File: tools/mail.py

#!/bin/bash
# This script has moved.
management/cli.py "$@"



File: tools/readable_bash.py

#!/usr/bin/python3
#
# Generate documentation for how this machine works by
# parsing our bash scripts!

import cgi, re
import markdown
from modgrammar import *

def generate_documentation():
	print("""<!DOCTYPE html>
<html>
    <head>
        <meta charset="utf-8">
        <meta http-equiv="X-UA-Compatible" content="IE=edge,chrome=1">
        <meta name="viewport" content="width=device-width">

        <title>Build Your Own Mail Server From Scratch</title>

        <link rel="stylesheet" href="https://maxcdn.bootstrapcdn.com/bootstrap/3.2.0/css/bootstrap.min.css">
        <link rel="stylesheet" href="https://maxcdn.bootstrapcdn.com/bootstrap/3.2.0/css/bootstrap-theme.min.css">

        <style>
		    @import url(https://fonts.googleapis.com/css?family=Iceland);
		    @import url(https://fonts.googleapis.com/css?family=Raleway:400,700);
			@import url(https://fonts.googleapis.com/css?family=Ubuntu:300,500);
		   	body {
		    		font-family: Raleway, sans-serif;
		    		font-size: 16px;
					color: #555;
	    	}
	    	h2, h3 {
	    		margin-top: .25em;
	    		margin-bottom: .75em;
	    	}
	    	p {
	    		margin-bottom: 1em;
	    	}
		    	.intro p {
		    		margin: 1.5em 0;
		    	}
		    li {
		    	margin-bottom: .33em;
		    }

			.sourcefile {
	    		padding-top: 1.5em;
	    		padding-bottom: 1em;
	    		font-size: 90%;
	    		text-align: right;
			}
				.sourcefile a {
					color: red;
				}

	    	.instructions .row.contd {
	    		border-top: 1px solid #E0E0E0;
	    	}

	    	.prose {
	    		padding-top: 1em;
	    		padding-bottom: 1em;
	    	}
	    	.terminal {
	    		background-color: #EEE;
	    		padding-top: 1em;
	    		padding-bottom: 1em;
	    	}

	    	ul {
	    		padding-left: 1.25em;
	    	}

	    	pre {
	    		color: black;
	    		border: 0;
	    		background: none;
	    		font-size: 100%;
	    	}

	    	div.write-to {
	    		margin: 0 0 1em .5em;
	    	}
	    	div.write-to p {
	    		padding: .5em;
	    		margin: 0;
	    	}
	    	div.write-to .filename {
	    		padding: .25em .5em;
	    		background-color: #666;
	    		color: white;
	    		font-family: monospace;
	    		font-weight: bold;
	    	}
	    	div.write-to .filename span {
	    		font-family: sans-serif;
	    		font-weight: normal;
	    	}
	    	div.write-to pre {
	    		margin: 0;
	    		padding: .5em;
	    		border: 1px solid #999;
	    		border-radius: 0;
	    		font-size: 90%;
	    	}

	    	pre.shell > div:before {
	    		content: "$ ";
	    		color: #666;
	    	}
        </style>
    </head>
    <body>
    <div class="container">
      <div class="row intro">
        <div class="col-xs-12">
        <h1>Build Your Own Mail Server From Scratch</h1>
        <p>Here&rsquo;s how you can build your own mail server from scratch.</p>
        <p>This document is generated automatically from <a href="https://mailinabox.email">Mail-in-a-Box</a>&rsquo;s setup script <a href="https://github.com/mail-in-a-box/mailinabox">source code</a>.</p>
        <hr>
      </div>
    </div>
    <div class="container instructions">
 """)

	parser = Source.parser()
	with open("setup/start.sh", "r") as start_file:
                for line in start_file:
                        try:
                                fn = parser.parse_string(line).filename()
                        except:
                                continue
                        if fn in ("setup/start.sh", "setup/preflight.sh", "setup/questions.sh", "setup/firstuser.sh", "setup/management.sh"):
                                continue

		import sys
		print(fn, file=sys.stderr)

		print(BashScript.parse(fn))

	print("""
        <script src="https://ajax.googleapis.com/ajax/libs/jquery/1.10.1/jquery.min.js"></script>
        <script src="https://maxcdn.bootstrapcdn.com/bootstrap/3.2.0/js/bootstrap.min.js"></script>
        <script>
        $(function() {
			$('.terminal').each(function() {
			  $(this).outerHeight( $(this).parent().innerHeight() );
			});
        })
		</script>
    </body>
</html>
""")

class HashBang(Grammar):
	grammar = (L('#!'), REST_OF_LINE, EOL)
	def value(self):
		return ""

def strip_indent(s):
	s = s.replace("\t", "    ")
	lines = s.split("\n")
	try:
		min_indent = min(len(re.match(r"\s*", line).group(0)) for line in lines if len(line) > 0)
	except ValueError:
		# No non-empty lines.
		min_indent = 0
	lines = [line[min_indent:] for line in lines]
	return "\n".join(lines)

class Comment(Grammar):
	grammar = ONE_OR_MORE(ZERO_OR_MORE(SPACE), L('#'), REST_OF_LINE, EOL)
	def value(self):
		if self.string.replace("#", "").strip() == "":
			return "\n"
		lines = [x[2].string for x in self[0]]
		content = "\n".join(lines)
		content = strip_indent(content)
		return markdown.markdown(content, output_format="html4") + "\n\n"

FILENAME = WORD('a-z0-9-/.')

class Source(Grammar):
	grammar = ((L('.') | L('source')), L(' '), FILENAME, Comment | EOL)
	def filename(self):
		return self[2].string.strip()
	def value(self):
		return BashScript.parse(self.filename())

class CatEOF(Grammar):
	grammar = (ZERO_OR_MORE(SPACE), L('cat '), L('>') | L('>>'), L(' '), ANY_EXCEPT(WHITESPACE), L(" <<"), OPTIONAL(SPACE), L("EOF"), EOL, REPEAT(ANY, greedy=False), EOL, L("EOF"), EOL)
	def value(self):
		content = self[9].string
		content = re.sub(r"\\([$])", r"\1", content) # un-escape bash-escaped characters
		return "<div class='write-to'><div class='filename'>%s <span>(%s)</span></div><pre>%s</pre></div>\n" \
			% (self[4].string,
			   "overwrite" if ">>" not in self[2].string else "append to",
			   cgi.escape(content))

class HideOutput(Grammar):
	grammar = (L("hide_output "), REF("BashElement"))
	def value(self):
		return self[1].value()

class EchoLine(Grammar):
	grammar = (OPTIONAL(SPACE), L("echo "), REST_OF_LINE, EOL)
	def value(self):
		if "|" in self.string  or ">" in self.string:
			return "<pre class='shell'><div>" + recode_bash(self.string.strip()) + "</div></pre>\n"
		return ""

class EditConf(Grammar):
	grammar = (
		L('tools/editconf.py '),
		FILENAME,
		SPACE,
		OPTIONAL((LIST_OF(
			L("-w") | L("-s") | L("-c ;"),
			sep=SPACE,
		), SPACE)),
		REST_OF_LINE,
		OPTIONAL(SPACE),
		EOL
		)
	def value(self):
		conffile = self[1]
		options = []
		eq = "="
		if self[3] and "-s" in self[3].string: eq = " "
		for opt in re.split("\s+", self[4].string):
			k, v = opt.split("=", 1)
			v = re.sub(r"\n+", "", fixup_tokens(v)) # not sure why newlines are getting doubled
			options.append("%s%s%s" % (k, eq, v))
		return "<div class='write-to'><div class='filename'>" + self[1].string + " <span>(change settings)</span></div><pre>" + "\n".join(cgi.escape(s) for s in options) + "</pre></div>\n"

class CaptureOutput(Grammar):
	grammar = OPTIONAL(SPACE), WORD("A-Za-z_"), L('=$('), REST_OF_LINE, L(")"), OPTIONAL(L(';')), EOL
	def value(self):
		cmd = self[3].string
		cmd = cmd.replace("; ", "\n")
		return "<div class='write-to'><div class='filename'>$" + self[1].string + "=</div><pre>" + cgi.escape(cmd) + "</pre></div>\n"

class SedReplace(Grammar):
	grammar = OPTIONAL(SPACE), L('sed -i "s/'), OPTIONAL(L('^')), ONE_OR_MORE(WORD("-A-Za-z0-9 #=\\{};.*$_!()")), L('/'), ONE_OR_MORE(WORD("-A-Za-z0-9 #=\\{};.*$_!()")), L('/"'), SPACE, FILENAME, EOL
	def value(self):
		return "<div class='write-to'><div class='filename'>edit<br>" + self[8].string + "</div><p>replace</p><pre>" + cgi.escape(self[3].string.replace(".*", ". . .")) + "</pre><p>with</p><pre>" + cgi.escape(self[5].string.replace("\\n", "\n").replace("\\t", "\t")) + "</pre></div>\n"

class EchoPipe(Grammar):
	grammar = OPTIONAL(SPACE), L("echo "), REST_OF_LINE, L(' | '), REST_OF_LINE, EOL
	def value(self):
		text = " ".join("\"%s\"" % s for s in self[2].string.split(" "))
		return "<pre class='shell'><div>echo " + recode_bash(text) + " \<br> | " + recode_bash(self[4].string) + "</div></pre>\n"

def shell_line(bash):
	return "<pre class='shell'><div>" + recode_bash(bash.strip()) + "</div></pre>\n"

class AptGet(Grammar):
	grammar = (ZERO_OR_MORE(SPACE), L("apt_install "), REST_OF_LINE, EOL)
	def value(self):
		return shell_line("apt-get install -y " + re.sub(r"\s+", " ", self[2].string))
class UfwAllow(Grammar):
	grammar = (ZERO_OR_MORE(SPACE), L("ufw_allow "), REST_OF_LINE, EOL)
	def value(self):
		return shell_line("ufw allow " + self[2].string)
class UfwLimit(Grammar):
	grammar = (ZERO_OR_MORE(SPACE), L("ufw_limit "), REST_OF_LINE, EOL)
	def value(self):
		return shell_line("ufw limit " + self[2].string)
class RestartService(Grammar):
	grammar = (ZERO_OR_MORE(SPACE), L("restart_service "), REST_OF_LINE, EOL)
	def value(self):
		return shell_line("service " + self[2].string + " restart")

class OtherLine(Grammar):
	grammar = (REST_OF_LINE, EOL)
	def value(self):
		if self.string.strip() == "": return ""
		if "source setup/functions.sh" in self.string: return ""
		if "source /etc/mailinabox.conf" in self.string: return ""
		return "<pre class='shell'><div>" + recode_bash(self.string.strip()) + "</div></pre>\n"

class BashElement(Grammar):
	grammar = Comment | CatEOF | EchoPipe | EchoLine | HideOutput | EditConf | SedReplace | AptGet | UfwAllow | UfwLimit | RestartService | OtherLine
	def value(self):
		return self[0].value()

# Make some special characters to private use Unicode code points.
bash_special_characters1 = {
	"\n": "\uE000",
	" ": "\uE001",
}
bash_special_characters2 = {
	"$": "\uE010",
}
bash_escapes = {
	"n": "\uE020",
	"t": "\uE021",
}

def quasitokenize(bashscript):
	# Make a parse of bash easier by making the tokenization easy.
	newscript = ""
	quote_mode = None
	escape_next = False
	line_comment = False
	subshell = 0
	for c in bashscript:
		if line_comment:
			# We're in a comment until the end of the line.
			newscript += c
			if c == '\n':
				line_comment = False
		elif escape_next:
			# Previous character was a \. Normally the next character
			# comes through literally, but escaped newlines are line
			# continuations and some escapes are for special characters
			# which we'll recode and then turn back into escapes later.
			if c == "\n":
				c = " "
			elif c in bash_escapes:
				c = bash_escapes[c]
			newscript += c
			escape_next = False
		elif c == "\\":
			# Escaping next character.
			escape_next = True
		elif quote_mode is None and c in ('"', "'"):
			# Starting a quoted word.
			quote_mode = c
		elif c == quote_mode:
			# Ending a quoted word.
			quote_mode = None
		elif quote_mode is not None and quote_mode != "EOF" and c in bash_special_characters1:
			# Replace special tokens within quoted words so that they
			# don't interfere with tokenization later.
			newscript += bash_special_characters1[c]
		elif quote_mode is None and c == '#':
			# Start of a line comment.
			newscript += c
			line_comment = True
		elif quote_mode is None and c == ';' and subshell == 0:
			# End of a statement.
			newscript += "\n"
		elif quote_mode is None and c == '(':
			# Start of a subshell.
			newscript += c
			subshell += 1
		elif quote_mode is None and c == ')':
			# End of a subshell.
			newscript += c
			subshell -= 1
		elif quote_mode is None and c == '\t':
			# Make these just spaces.
			if newscript[-1] != " ":
				newscript += " "
		elif quote_mode is None and c == ' ':
			# Collapse consecutive spaces.
			if newscript[-1] != " ":
				newscript += " "
		elif c in bash_special_characters2:
			newscript += bash_special_characters2[c]
		else:
			# All other characters.
			newscript += c

		# "<< EOF" escaping.
		if quote_mode is None and re.search("<<\s*EOF\n$", newscript):
			quote_mode = "EOF"
		elif quote_mode == "EOF" and re.search("\nEOF\n$", newscript):
			quote_mode = None

	return newscript

def recode_bash(s):
	def requote(tok):
		tok = tok.replace("\\", "\\\\")
		for c in bash_special_characters2:
			tok = tok.replace(c, "\\" + c)
		tok = fixup_tokens(tok)
		if " " in tok or '"' in tok:
			tok = tok.replace("\"", "\\\"")
			tok = '"' + tok +'"'
		else:
			tok = tok.replace("'", "\\'")
		return tok
	return cgi.escape(" ".join(requote(tok) for tok in s.split(" ")))

def fixup_tokens(s):
	for c, enc in bash_special_characters1.items():
		s = s.replace(enc, c)
	for c, enc in bash_special_characters2.items():
		s = s.replace(enc, c)
	for esc, c in bash_escapes.items():
		s = s.replace(c, "\\" + esc)
	return s

class BashScript(Grammar):
	grammar = (OPTIONAL(HashBang), REPEAT(BashElement))
	def value(self):
		return [line.value() for line in self[1]]

	@staticmethod
	def parse(fn):
		if fn in ("setup/functions.sh", "/etc/mailinabox.conf"): return ""
		with open(fn, "r") as f:
			string = f.read()

		# tokenize
		string = re.sub(".* #NODOC\n", "", string)
		string = re.sub("\n\s*if .*then.*|\n\s*fi|\n\s*else|\n\s*elif .*", "", string)
		string = quasitokenize(string)
		string = re.sub("hide_output ", "", string)

		parser = BashScript.parser()
		result = parser.parse_string(string)

		v = "<div class='row'><div class='col-xs-12 sourcefile'>view the bash source for the following section at <a href=\"%s\">%s</a></div></div>\n" \
			 % ("https://github.com/mail-in-a-box/mailinabox/tree/master/" + fn, fn)

		mode = 0
		for item in result.value():
			if item.strip() == "":
				pass
			elif item.startswith("<p") and not item.startswith("<pre"):
				clz = ""
				if mode == 2:
					v += "</div>\n" # col
					v += "</div>\n" # row
					mode = 0
					clz = "contd"
				if mode == 0:
					v += "<div class='row %s'>\n" % clz
					v += "<div class='col-md-6 prose'>\n"
				v += item
				mode = 1
			elif item.startswith("<h"):
				if mode != 0:
					v += "</div>\n" # col
					v += "</div>\n" # row
				v += "<div class='row'>\n"
				v += "<div class='col-md-6 header'>\n"
				v += item
				v += "</div>\n" # col
				v += "<div class='col-md-6 terminal'> </div>\n"
				v += "</div>\n" # row
				mode = 0
			else:
				if mode == 0:
					v += "<div class='row'>\n"
					v += "<div class='col-md-offset-6 col-md-6 terminal'>\n"
				elif mode == 1:
					v += "</div>\n"
					v += "<div class='col-md-6 terminal'>\n"
				mode = 2
				v += item

		v += "</div>\n" # col
		v += "</div>\n" # row

		v = fixup_tokens(v)

		v = v.replace("</pre>\n<pre class='shell'>", "")
		v = re.sub("<pre>([\w\W]*?)</pre>", lambda m : "<pre>" + strip_indent(m.group(1)) + "</pre>", v)

		v = re.sub(r"(\$?)PRIMARY_HOSTNAME", r"<b>box.yourdomain.com</b>", v)
		v = re.sub(r"\$STORAGE_ROOT", r"<b>$STORE</b>", v)
		v = v.replace("`pwd`",  "<code><b>/path/to/mailinabox</b></code>")

		return v

def wrap_lines(text, cols=60):
	ret = ""
	words = re.split("(\s+)", text)
	linelen = 0
	for w in words:
		if linelen + len(w) > cols-1:
			ret += " \\\n"
			ret += "   "
			linelen = 0
		if linelen == 0 and w.strip() == "": continue
		ret += w
		linelen += len(w)
	return ret

if __name__ == '__main__':
	generate_documentation()



File: tools/parse-nginx-log-bootstrap-accesses.py

#!/usr/bin/python3
#
# This is a tool Josh uses on his box serving mailinabox.email to parse the nginx
# access log to see how many people are installing Mail-in-a-Box each day, by
# looking at accesses to the bootstrap.sh script (which is currently at the URL
# .../setup.sh).

import re, glob, gzip, os.path, json
import dateutil.parser

outfn = "/home/user-data/www/mailinabox.email/install-stats.json"

# Make a unique list of (date, ip address) pairs so we don't double-count
# accesses that are for the same install.
accesses = set()

# Scan the current and rotated access logs.
for fn in glob.glob("/var/log/nginx/access.log*"):
	# Gunzip if necessary.
	# Loop through the lines in the access log.
	with (gzip.open if fn.endswith(".gz") else open)(fn, "rb") as f:
		for line in f:
			# Find lines that are GETs on the bootstrap script by either curl or wget.
			# (Note that we purposely skip ...?ping=1 requests which is the admin panel querying us for updates.)
			# (Also, the URL changed in January 2016, but we'll accept both.)
			m = re.match(rb"(?P<ip>\S+) - - \[(?P<date>.*?)\] \"GET /(bootstrap.sh|setup.sh) HTTP/.*\" 200 \d+ .* \"(?:curl|wget)", line, re.I)
			if m:
				date, time = m.group("date").decode("ascii").split(":", 1)
				date = dateutil.parser.parse(date).date().isoformat()
				ip = m.group("ip").decode("ascii")
				accesses.add( (date, ip) )

# Aggregate by date.
by_date = { }
for date, ip in accesses:
	by_date[date] = by_date.get(date, 0) + 1

# Since logs are rotated, store the statistics permanently in a JSON file.
# Load in the stats from an existing file.
if os.path.exists(outfn):
	with open(outfn, encoding="utf-8") as f:
		existing_data = json.load(f)
	for date, count in existing_data:
		if date not in by_date:
			by_date[date] = count

# Turn into a list rather than a dict structure to make it ordered.
by_date = sorted(by_date.items())

# Pop the last one because today's stats are incomplete.
by_date.pop(-1)

# Write out.
with open(outfn, "w", encoding="utf-8") as f:
	json.dump(by_date, f, sort_keys=True, indent=True)



File: tools/editconf.py

#!/usr/bin/python3
#
# This is a helper tool for editing configuration files during the setup
# process. The tool is given new values for settings as command-line
# arguments. It comments-out existing setting values in the configuration
# file and adds new values either after their former location or at the
# end.
#
# The configuration file has settings that look like:
#
# NAME=VALUE
#
# If the -s option is given, then space becomes the delimiter, i.e.:
#
# NAME VALUE
#
# If the -e option is given and VALUE is empty, the setting is removed
# from the configuration file if it is set (i.e. existing occurrences
# are commented out and no new setting is added).
#
# If the -c option is given, then the supplied character becomes the comment character
#
# If the -w option is given, then setting lines continue onto following
# lines while the lines start with whitespace, e.g.:
#
# NAME VAL
#   UE

import sys, re

# sanity check
if len(sys.argv) < 3:
	print("usage: python3 editconf.py /etc/file.conf [-e] [-s] [-w] [-c <CHARACTER>] [-t] NAME=VAL [NAME=VAL ...]")
	sys.exit(1)

# parse command line arguments
filename = sys.argv[1]
settings = sys.argv[2:]

delimiter = "="
delimiter_re = r"\s*=\s*"
erase_setting = False
comment_char = "#"
folded_lines = False
testing = False
while settings[0][0] == "-" and settings[0] != "--":
	opt = settings.pop(0)
	if opt == "-s":
		# Space is the delimiter
		delimiter = " "
		delimiter_re = r"\s+"
	elif opt == "-e":
		# Erase settings that have empty values.
		erase_setting = True
	elif opt == "-w":
		# Line folding is possible in this file.
		folded_lines = True
	elif opt == "-c":
		# Specifies a different comment character.
		comment_char = settings.pop(0)
	elif opt == "-t":
		testing = True
	else:
		print("Invalid option.")
		sys.exit(1)

# sanity check command line
for setting in settings:
	try:
		name, value = setting.split("=", 1)
	except:
		import subprocess
		print("Invalid command line: ", subprocess.list2cmdline(sys.argv))

# create the new config file in memory

found = set()
buf = ""
with open(filename, encoding="utf-8") as f:
        input_lines = list(f)

while len(input_lines) > 0:
	line = input_lines.pop(0)

	# If this configuration file uses folded lines, append any folded lines
	# into our input buffer.
	if folded_lines and line[0] not in {comment_char, " ", ""}:
		while len(input_lines) > 0 and input_lines[0][0] in " \t":
			line += input_lines.pop(0)

	# See if this line is for any settings passed on the command line.
	for i in range(len(settings)):
		# Check if this line contain this setting from the command-line arguments.
		name, val = settings[i].split("=", 1)
		m = re.match(
			   r"(\s*)"
			 "(" + re.escape(comment_char) + r"\s*)?"
			 + re.escape(name) + delimiter_re + r"(.*?)\s*$",
			 line, re.S)
		if not m: continue
		indent, is_comment, existing_val = m.groups()

		# If this is already the setting, keep it in the file, except:
		# * If we've already seen it before, then remove this duplicate line.
		# * If val is empty and erase_setting is on, then comment it out.
		if is_comment is None and existing_val == val and not (not val and erase_setting):
			# It may be that we've already inserted this setting higher
			# in the file so check for that first.
			if i in found: break
			buf += line
			found.add(i)
			break

		# comment-out the existing line (also comment any folded lines)
		if is_comment is None:
			buf += comment_char + line.rstrip().replace("\n", "\n" + comment_char) + "\n"
		else:
			# the line is already commented, pass it through
			buf += line

		# if this option already is set don't add the setting again,
		# or if we're clearing the setting with -e, don't add it
		if (i in found) or (not val and erase_setting):
			break

		# add the new setting
		buf += indent + name + delimiter + val + "\n"

		# note that we've applied this option
		found.add(i)

		break
	else:
		# If did not match any setting names, pass this line through.
		buf += line

# Put any settings we didn't see at the end of the file,
# except settings being cleared.
for i in range(len(settings)):
	if i not in found:
		name, val = settings[i].split("=", 1)
		if not (not val and erase_setting):
			buf += name + delimiter + val + "\n"

if not testing:
	# Write out the new file.
	with open(filename, "w", encoding="utf-8") as f:
		f.write(buf)
else:
	# Just print the new file to stdout.
	print(buf)



File: setup/migrate.py

#!/usr/bin/python3

# Migrates any file structures, database schemas, etc. between versions of Mail-in-a-Box.

# We have to be careful here that any dependencies are already installed in the previous
# version since this script runs before all other aspects of the setup script.

import sys, os, os.path, glob, re, shutil

sys.path.insert(0, 'management')
from utils import load_environment, save_environment, shell
import contextlib

def migration_1(env):
	# Re-arrange where we store SSL certificates. There was a typo also.

	def move_file(fn, domain_name_escaped, filename):
		# Moves an SSL-related file into the right place.
		fn1 = os.path.join( env["STORAGE_ROOT"], 'ssl', domain_name_escaped, file_type)
		os.makedirs(os.path.dirname(fn1), exist_ok=True)
		shutil.move(fn, fn1)

	# Migrate the 'domains' directory.
	for sslfn in glob.glob(os.path.join( env["STORAGE_ROOT"], 'ssl/domains/*' )):
		fn = os.path.basename(sslfn)
		m = re.match("(.*)_(certifiate.pem|cert_sign_req.csr|private_key.pem)$", fn)
		if m:
			# get the new name for the file
			domain_name, file_type = m.groups()
			if file_type == "certifiate.pem": file_type = "ssl_certificate.pem" # typo
			if file_type == "cert_sign_req.csr": file_type = "certificate_signing_request.csr" # nicer
			move_file(sslfn, domain_name, file_type)

	# Move the old domains directory if it is now empty.
	with contextlib.suppress(Exception):
		os.rmdir(os.path.join( env["STORAGE_ROOT"], 'ssl/domains'))

def migration_2(env):
	# Delete the .dovecot_sieve script everywhere. This was formerly a copy of our spam -> Spam
	# script. We now install it as a global script, and we use managesieve, so the old file is
	# irrelevant. Also delete the compiled binary form.
	for fn in glob.glob(os.path.join(env["STORAGE_ROOT"], 'mail/mailboxes/*/*/.dovecot.sieve')):
		os.unlink(fn)
	for fn in glob.glob(os.path.join(env["STORAGE_ROOT"], 'mail/mailboxes/*/*/.dovecot.svbin')):
		os.unlink(fn)

def migration_3(env):
	# Move the migration ID from /etc/mailinabox.conf to $STORAGE_ROOT/mailinabox.version
	# so that the ID stays with the data files that it describes the format of. The writing
	# of the file will be handled by the main function.
	pass

def migration_4(env):
	# Add a new column to the mail users table where we can store administrative privileges.
	db = os.path.join(env["STORAGE_ROOT"], 'mail/users.sqlite')
	shell("check_call", ["sqlite3", db, "ALTER TABLE users ADD privileges TEXT NOT NULL DEFAULT ''"])

def migration_5(env):
	# The secret key for encrypting backups was world readable. Fix here.
	os.chmod(os.path.join(env["STORAGE_ROOT"], 'backup/secret_key.txt'), 0o600)

def migration_6(env):
	# We now will generate multiple DNSSEC keys for different algorithms, since TLDs may
	# not support them all. .email only supports RSA/SHA-256. Rename the keys.conf file
	# to be algorithm-specific.
	basepath = os.path.join(env["STORAGE_ROOT"], 'dns/dnssec')
	shutil.move(os.path.join(basepath, 'keys.conf'), os.path.join(basepath, 'RSASHA1-NSEC3-SHA1.conf'))

def migration_7(env):
	# I previously wanted domain names to be stored in Unicode in the database. Now I want them
	# to be in IDNA. Affects aliases only.
	import sqlite3
	conn = sqlite3.connect(os.path.join(env["STORAGE_ROOT"], "mail/users.sqlite"))

	# Get existing alias source addresses.
	c = conn.cursor()
	c.execute('SELECT source FROM aliases')
	aliases = [ row[0] for row in c.fetchall() ]

	# Update to IDNA-encoded domains.
	for email in aliases:
		try:
			localpart, domainpart = email.split("@")
			domainpart = domainpart.encode("idna").decode("ascii")
			newemail = localpart + "@" + domainpart
			if newemail != email:
				c = conn.cursor()
				c.execute("UPDATE aliases SET source=? WHERE source=?", (newemail, email))
				if c.rowcount != 1: raise ValueError("Alias not found.")
				print("Updated alias", email, "to", newemail)
		except Exception as e:
			print("Error updating IDNA alias", email, e)

	# Save.
	conn.commit()

def migration_8(env):
	# Delete DKIM keys. We had generated 1024-bit DKIM keys.
	# By deleting the key file we'll automatically generate
	# a new key, which will be 2048 bits.
	os.unlink(os.path.join(env['STORAGE_ROOT'], 'mail/dkim/mail.private'))

def migration_9(env):
	# Add a column to the aliases table to store permitted_senders,
	# which is a list of user account email addresses that are
	# permitted to send mail using this alias instead of their own
	# address. This was motivated by the addition of #427 ("Reject
	# outgoing mail if FROM does not match Login") - which introduced
	# the notion of outbound permitted-senders.
	db = os.path.join(env["STORAGE_ROOT"], 'mail/users.sqlite')
	shell("check_call", ["sqlite3", db, "ALTER TABLE aliases ADD permitted_senders TEXT"])

def migration_10(env):
	# Clean up the SSL certificates directory.

	# Move the primary certificate to a new name and then
	# symlink it to the system certificate path.
	import datetime
	system_certificate = os.path.join(env["STORAGE_ROOT"], 'ssl/ssl_certificate.pem')
	if not os.path.islink(system_certificate): # not already a symlink
		new_path = os.path.join(env["STORAGE_ROOT"], 'ssl', env['PRIMARY_HOSTNAME'] + "-" + datetime.datetime.now().date().isoformat().replace("-", "") + ".pem")
		print("Renamed", system_certificate, "to", new_path, "and created a symlink for the original location.")
		shutil.move(system_certificate, new_path)
		os.symlink(new_path, system_certificate)

	# Flatten the directory structure. For any directory
	# that contains a single file named ssl_certificate.pem,
	# move the file out and name it the same as the directory,
	# and remove the directory.
	for sslcert in glob.glob(os.path.join( env["STORAGE_ROOT"], 'ssl/*/ssl_certificate.pem' )):
		d = os.path.dirname(sslcert)
		if len(os.listdir(d)) == 1:
			# This certificate is the only file in that directory.
			newname = os.path.join(env["STORAGE_ROOT"], 'ssl', os.path.basename(d) + '.pem')
			if not os.path.exists(newname):
				shutil.move(sslcert, newname)
				os.rmdir(d)

def migration_11(env):
	# Archive the old Let's Encrypt account directory managed by free_tls_certificates
	# because we'll use that path now for the directory managed by certbot.
	try:
		old_path = os.path.join(env["STORAGE_ROOT"], 'ssl', 'lets_encrypt')
		new_path = os.path.join(env["STORAGE_ROOT"], 'ssl', 'lets_encrypt-old')
		shutil.move(old_path, new_path)
	except:
		# meh
		pass

def migration_12(env):
	# Upgrading to Carddav Roundcube plugin to version 3+, it requires the carddav_*
        # tables to be dropped.
        # Checking that the roundcube database already exists.
        if os.path.exists(os.path.join(env["STORAGE_ROOT"], "mail/roundcube/roundcube.sqlite")):
            import sqlite3
            conn = sqlite3.connect(os.path.join(env["STORAGE_ROOT"], "mail/roundcube/roundcube.sqlite"))
            c = conn.cursor()
            # Get a list of all the tables that begin with 'carddav_'
            c.execute("SELECT name FROM sqlite_master WHERE type = ? AND name LIKE ?", ('table', 'carddav_%'))
            carddav_tables = c.fetchall()
            # If there were tables that begin with 'carddav_', drop them
            if carddav_tables:
                for table in carddav_tables:
                    try:
                        table = table[0]
                        c = conn.cursor()
                        dropcmd = "DROP TABLE %s" % table
                        c.execute(dropcmd)
                    except:
                        print("Failed to drop table", table)
            # Save.
            conn.commit()
            conn.close()

            # Delete all sessions, requiring users to login again to recreate carddav_*
            # databases
            conn = sqlite3.connect(os.path.join(env["STORAGE_ROOT"], "mail/roundcube/roundcube.sqlite"))
            c = conn.cursor()
            c.execute("delete from session;")
            conn.commit()
            conn.close()

def migration_13(env):
	# Add the "mfa" table for configuring MFA for login to the control panel.
	db = os.path.join(env["STORAGE_ROOT"], 'mail/users.sqlite')
	shell("check_call", ["sqlite3", db, "CREATE TABLE mfa (id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, type TEXT NOT NULL, secret TEXT NOT NULL, mru_token TEXT, label TEXT, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE);"])

def migration_14(env):
	# Add the "auto_aliases" table.
	db = os.path.join(env["STORAGE_ROOT"], 'mail/users.sqlite')
	shell("check_call", ["sqlite3", db, "CREATE TABLE auto_aliases (id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT NOT NULL UNIQUE, destination TEXT NOT NULL, permitted_senders TEXT);"])

###########################################################

def get_current_migration():
	ver = 0
	while True:
		next_ver = (ver + 1)
		migration_func = globals().get("migration_%d" % next_ver)
		if not migration_func:
			return ver
		ver = next_ver

def run_migrations():
	if not os.access("/etc/mailinabox.conf", os.W_OK, effective_ids=True):
		print("This script must be run as root.", file=sys.stderr)
		sys.exit(1)

	env = load_environment()

	migration_id_file = os.path.join(env['STORAGE_ROOT'], 'mailinabox.version')
	migration_id = None
	if os.path.exists(migration_id_file):
		with open(migration_id_file, encoding='utf-8') as f:
			migration_id = f.read().strip()

	if migration_id is None:
		# Load the legacy location of the migration ID. We'll drop support
		# for this eventually.
		migration_id = env.get("MIGRATIONID")

	if migration_id is None:
		print()
		print(f"{migration_id_file} file doesn't exists. Skipping migration...")
		return

	ourver = int(migration_id)

	while True:
		next_ver = (ourver + 1)
		migration_func = globals().get("migration_%d" % next_ver)

		if not migration_func:
			# No more migrations to run.
			break

		print()
		print("Running migration to Mail-in-a-Box #%d..." % next_ver)

		try:
			migration_func(env)
		except Exception as e:
			print()
			print("Error running the migration script:")
			print()
			print(e)
			print()
			print("Your system may be in an inconsistent state now. We're terribly sorry. A re-install from a backup might be the best way to continue.")
			sys.exit(1)

		ourver = next_ver

		# Write out our current version now. Do this sooner rather than later
		# in case of any problems.
		with open(migration_id_file, "w", encoding='utf-8') as f:
			f.write(str(ourver) + "\n")

		# Delete the legacy location of this field.
		if "MIGRATIONID" in env:
			del env["MIGRATIONID"]
			save_environment(env)

		# iterate and try next version...

if __name__ == "__main__":
	if sys.argv[-1] == "--current":
		# Return the number of the highest migration.
		print(str(get_current_migration()))
	elif sys.argv[-1] == "--migrate":
		# Perform migrations.
		run_migrations()




File: management/mfa.py

import base64
import hmac
import io
import os
import pyotp
import qrcode

from mailconfig import open_database

def get_user_id(email, c):
	c.execute('SELECT id FROM users WHERE email=?', (email,))
	r = c.fetchone()
	if not r: raise ValueError("User does not exist.")
	return r[0]

def get_mfa_state(email, env):
	c = open_database(env)
	c.execute('SELECT id, type, secret, mru_token, label FROM mfa WHERE user_id=?', (get_user_id(email, c),))
	return [
		{ "id": r[0], "type": r[1], "secret": r[2], "mru_token": r[3], "label": r[4] }
		for r in c.fetchall()
	]

def get_public_mfa_state(email, env):
	mfa_state = get_mfa_state(email, env)
	return [
		{ "id": s["id"], "type": s["type"], "label": s["label"] }
		for s in mfa_state
	]

def get_hash_mfa_state(email, env):
	mfa_state = get_mfa_state(email, env)
	return [
		{ "id": s["id"], "type": s["type"], "secret": s["secret"] }
		for s in mfa_state
	]

def enable_mfa(email, type, secret, token, label, env):
	if type == "totp":
		validate_totp_secret(secret)
		# Sanity check with the provide current token.
		totp = pyotp.TOTP(secret)
		if not totp.verify(token, valid_window=1):
			msg = "Invalid token."
			raise ValueError(msg)
	else:
		msg = "Invalid MFA type."
		raise ValueError(msg)

	conn, c = open_database(env, with_connection=True)
	c.execute('INSERT INTO mfa (user_id, type, secret, label) VALUES (?, ?, ?, ?)', (get_user_id(email, c), type, secret, label))
	conn.commit()

def set_mru_token(email, mfa_id, token, env):
	conn, c = open_database(env, with_connection=True)
	c.execute('UPDATE mfa SET mru_token=? WHERE user_id=? AND id=?', (token, get_user_id(email, c), mfa_id))
	conn.commit()

def disable_mfa(email, mfa_id, env):
	conn, c = open_database(env, with_connection=True)
	if mfa_id is None:
		# Disable all MFA for a user.
		c.execute('DELETE FROM mfa WHERE user_id=?', (get_user_id(email, c),))
	else:
		# Disable a particular MFA mode for a user.
		c.execute('DELETE FROM mfa WHERE user_id=? AND id=?', (get_user_id(email, c), mfa_id))
	conn.commit()
	return c.rowcount > 0

def validate_totp_secret(secret):
	if not isinstance(secret, str) or secret.strip() == "":
		msg = "No secret provided."
		raise ValueError(msg)
	if len(secret) != 32:
		msg = "Secret should be a 32 characters base32 string"
		raise ValueError(msg)

def provision_totp(email, env):
	# Make a new secret.
	secret = base64.b32encode(os.urandom(20)).decode('utf-8')
	validate_totp_secret(secret) # sanity check

	# Make a URI that we encode within a QR code.
	uri = pyotp.TOTP(secret).provisioning_uri(
		name=email,
		issuer_name=env["PRIMARY_HOSTNAME"] + " Mail-in-a-Box Control Panel"
	)

	# Generate a QR code as a base64-encode PNG image.
	qr = qrcode.make(uri)
	byte_arr = io.BytesIO()
	qr.save(byte_arr, format='PNG')
	png_b64 = base64.b64encode(byte_arr.getvalue()).decode('utf-8')

	return {
		"type": "totp",
		"secret": secret,
		"qr_code_base64": png_b64
	}

def validate_auth_mfa(email, request, env):
	# Validates that a login request satisfies any MFA modes
	# that have been enabled for the user's account. Returns
	# a tuple (status, [hints]). status is True for a successful
	# MFA login, False for a missing token. If status is False,
	# hints is an array of codes that indicate what the user
	# can try. Possible codes are:
	# "missing-totp-token"
	# "invalid-totp-token"

	mfa_state = get_mfa_state(email, env)

	# If no MFA modes are added, return True.
	if len(mfa_state) == 0:
		return (True, [])

	# Try the enabled MFA modes.
	hints = set()
	for mfa_mode in mfa_state:
		if mfa_mode["type"] == "totp":
			# Check that a token is present in the X-Auth-Token header.
			# If not, give a hint that one can be supplied.
			token = request.headers.get('x-auth-token')
			if not token:
				hints.add("missing-totp-token")
				continue

			# Check for a replay attack.
			if hmac.compare_digest(token, mfa_mode['mru_token'] or ""):
				# If the token fails, skip this MFA mode.
				hints.add("invalid-totp-token")
				continue

			# Check the token.
			totp = pyotp.TOTP(mfa_mode["secret"])
			if not totp.verify(token, valid_window=1):
				hints.add("invalid-totp-token")
				continue

			# On success, record the token to prevent a replay attack.
			set_mru_token(email, mfa_mode['id'], token, env)
			return (True, [])

	# On a failed login, indicate failure and any hints for what the user can do instead.
	return (False, list(hints))



File: management/auth.py

import base64, hmac, json, secrets
from datetime import timedelta

from expiringdict import ExpiringDict

import utils
from mailconfig import get_mail_password, get_mail_user_privileges
from mfa import get_hash_mfa_state, validate_auth_mfa

DEFAULT_KEY_PATH   = '/var/lib/mailinabox/api.key'
DEFAULT_AUTH_REALM = 'Mail-in-a-Box Management Server'

class AuthService:
	def __init__(self):
		self.auth_realm = DEFAULT_AUTH_REALM
		self.key_path = DEFAULT_KEY_PATH
		self.max_session_duration = timedelta(days=2)

		self.init_system_api_key()
		self.sessions = ExpiringDict(max_len=64, max_age_seconds=self.max_session_duration.total_seconds())

	def init_system_api_key(self):
		"""Write an API key to a local file so local processes can use the API"""

		with open(self.key_path, encoding='utf-8') as file:
			self.key = file.read()

	def authenticate(self, request, env, login_only=False, logout=False):
		"""Test if the HTTP Authorization header's username matches the system key, a session key,
		or if the username/password passed in the header matches a local user.
		Returns a tuple of the user's email address and list of user privileges (e.g.
		('my@email', []) or ('my@email', ['admin']); raises a ValueError on login failure.
		If the user used the system API key, the user's email is returned as None since
		this key is not associated with a user."""

		def parse_http_authorization_basic(header):
			def decode(s):
				return base64.b64decode(s.encode('ascii')).decode('ascii')
			if " " not in header:
				return None, None
			scheme, credentials = header.split(maxsplit=1)
			if scheme != 'Basic':
				return None, None
			credentials = decode(credentials)
			if ":" not in credentials:
				return None, None
			username, password = credentials.split(':', maxsplit=1)
			return username, password

		username, password = parse_http_authorization_basic(request.headers.get('Authorization', ''))
		if username in {None, ""}:
			msg = "Authorization header invalid."
			raise ValueError(msg)

		if username.strip() == "" and password.strip() == "":
			msg = "No email address, password, session key, or API key provided."
			raise ValueError(msg)

		# If user passed the system API key, grant administrative privs. This key
		# is not associated with a user.
		if username == self.key and not login_only:
			return (None, ["admin"])

		# If the password corresponds with a session token for the user, grant access for that user.
		if self.get_session(username, password, "login", env) and not login_only:
			sessionid = password
			session = self.sessions[sessionid]
			if logout:
				# Clear the session.
				del self.sessions[sessionid]
			else:
				# Re-up the session so that it does not expire.
				self.sessions[sessionid] = session

		# If no password was given, but a username was given, we're missing some information.
		elif password.strip() == "":
			msg = "Enter a password."
			raise ValueError(msg)

		else:
			# The user is trying to log in with a username and a password
			# (and possibly a MFA token). On failure, an exception is raised.
			self.check_user_auth(username, password, request, env)

		# Get privileges for authorization. This call should never fail because by this
		# point we know the email address is a valid user --- unless the user has been
		# deleted after the session was granted. On error the call will return a tuple
		# of an error message and an HTTP status code.
		privs = get_mail_user_privileges(username, env)
		if isinstance(privs, tuple): raise ValueError(privs[0])

		# Return the authorization information.
		return (username, privs)

	def check_user_auth(self, email, pw, request, env):
		# Validate a user's login email address and password. If MFA is enabled,
		# check the MFA token in the X-Auth-Token header.
		#
		# On login failure, raises a ValueError with a login error message. On
		# success, nothing is returned.

		# Authenticate.
		try:
			# Get the hashed password of the user. Raise a ValueError if the
			# email address does not correspond to a user. But wrap it in the
			# same exception as if a password fails so we don't easily reveal
			# if an email address is valid.
			pw_hash = get_mail_password(email, env)

			# Use 'doveadm pw' to check credentials. doveadm will return
			# a non-zero exit status if the credentials are no good,
			# and check_call will raise an exception in that case.
			utils.shell('check_call', [
				"/usr/bin/doveadm", "pw",
				"-p", pw,
				"-t", pw_hash,
				])
		except:
			# Login failed.
			msg = "Incorrect email address or password."
			raise ValueError(msg)

		# If MFA is enabled, check that MFA passes.
		status, hints = validate_auth_mfa(email, request, env)
		if not status:
			# Login valid. Hints may have more info.
			raise ValueError(",".join(hints))

	def create_user_password_state_token(self, email, env):
		# Create a token that changes if the user's password or MFA options change
		# so that sessions become invalid if any of that information changes.
		msg = get_mail_password(email, env).encode("utf8")

		# Add to the message the current MFA state, which is a list of MFA information.
		# Turn it into a string stably.
		msg += b" " + json.dumps(get_hash_mfa_state(email, env), sort_keys=True).encode("utf8")

		# Make a HMAC using the system API key as a hash key.
		hash_key = self.key.encode('ascii')
		return hmac.new(hash_key, msg, digestmod="sha256").hexdigest()

	def create_session_key(self, username, env, type=None):
		# Create a new session.
		token = secrets.token_hex(32)
		self.sessions[token] = {
			"email": username,
			"password_token": self.create_user_password_state_token(username, env),
			"type": type,
		}
		return token

	def get_session(self, user_email, session_key, session_type, env):
		if session_key not in self.sessions: return None
		session = self.sessions[session_key]
		if session_type == "login" and session["email"] != user_email: return None
		if session["type"] != session_type: return None
		if session["password_token"] != self.create_user_password_state_token(session["email"], env): return None
		return session



File: management/ssl_certificates.py

#!/usr/local/lib/mailinabox/env/bin/python
# Utilities for installing and selecting SSL certificates.

import os, os.path, re, shutil, subprocess, tempfile

from utils import shell, safe_domain_name, sort_domains
import functools
import operator

# SELECTING SSL CERTIFICATES FOR USE IN WEB

def get_ssl_certificates(env):
	# Scan all of the installed SSL certificates and map every domain
	# that the certificates are good for to the best certificate for
	# the domain.

	from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
	from cryptography.x509 import Certificate

	# The certificates are all stored here:
	ssl_root = os.path.join(env["STORAGE_ROOT"], 'ssl')

	# List all of the files in the SSL directory and one level deep.
	def get_file_list():
		if not os.path.exists(ssl_root):
			return
		for fn in os.listdir(ssl_root):
			if fn == 'ssl_certificate.pem':
				# This is always a symbolic link
				# to the certificate to use for
				# PRIMARY_HOSTNAME. Don't let it
				# be eligible for use because we
				# could end up creating a symlink
				# to itself --- we want to find
				# the cert that it should be a
				# symlink to.
				continue
			fn = os.path.join(ssl_root, fn)
			if os.path.isfile(fn):
				yield fn
			elif os.path.isdir(fn):
				for fn1 in os.listdir(fn):
					fn1 = os.path.join(fn, fn1)
					if os.path.isfile(fn1):
						yield fn1

	# Remember stuff.
	private_keys = { }
	certificates = [ ]

	# Scan each of the files to find private keys and certificates.
	# We must load all of the private keys first before processing
	# certificates so that we can check that we have a private key
	# available before using a certificate.
	for fn in get_file_list():
		try:
			pem = load_pem(load_cert_chain(fn)[0])
		except ValueError:
			# Not a valid PEM format for a PEM type we care about.
			continue

		# Is it a private key?
		if isinstance(pem, RSAPrivateKey):
			private_keys[pem.public_key().public_numbers()] = { "filename": fn, "key": pem }

		# Is it a certificate?
		if isinstance(pem, Certificate):
			certificates.append({ "filename": fn, "cert": pem })

	# Process the certificates.
	domains = { }
	for cert in certificates:
		# What domains is this certificate good for?
		cert_domains, primary_domain = get_certificate_domains(cert["cert"])
		cert["primary_domain"] = primary_domain

		# Is there a private key file for this certificate?
		private_key = private_keys.get(cert["cert"].public_key().public_numbers())
		if not private_key:
			continue
		cert["private_key"] = private_key

		# Add this cert to the list of certs usable for the domains.
		for domain in cert_domains:
			# The primary hostname can only use a certificate mapped
			# to the system private key.
			if domain == env['PRIMARY_HOSTNAME'] and cert["private_key"]["filename"] != os.path.join(env['STORAGE_ROOT'], 'ssl', 'ssl_private_key.pem'):
				continue

			domains.setdefault(domain, []).append(cert)

	# Sort the certificates to prefer good ones.
	import datetime
	now = datetime.datetime.utcnow()
	ret = { }
	for domain, cert_list in domains.items():
		#for c in cert_list: print(domain, c.not_valid_before, c.not_valid_after, "("+str(now)+")", c.issuer, c.subject, c._filename)
		cert_list.sort(key = lambda cert : (
			# must be valid NOW
			cert["cert"].not_valid_before <= now <= cert["cert"].not_valid_after,

			# prefer one that is not self-signed
			cert["cert"].issuer != cert["cert"].subject,

			###########################################################
			# The above lines ensure that valid certificates are chosen
			# over invalid certificates. The lines below choose between
			# multiple valid certificates available for this domain.
			###########################################################

			# prefer one with the expiration furthest into the future so
			# that we can easily rotate to new certs as we get them
			cert["cert"].not_valid_after,

			###########################################################
			# We always choose the certificate that is good for the
			# longest period of time. This is important for how we
			# provision certificates for Let's Encrypt. To ensure that
			# we don't re-provision every night, we have to ensure that
			# if we choose to provison a certificate that it will
			# *actually* be used so the provisioning logic knows it
			# doesn't still need to provision a certificate for the
			# domain.
			###########################################################

			# in case a certificate is installed in multiple paths,
			# prefer the... lexicographically last one?
			cert["filename"],

		), reverse=True)
		cert = cert_list.pop(0)
		ret[domain] = {
			"private-key": cert["private_key"]["filename"],
			"certificate": cert["filename"],
			"primary-domain": cert["primary_domain"],
			"certificate_object": cert["cert"],
			}

	return ret

def get_domain_ssl_files(domain, ssl_certificates, env, allow_missing_cert=False, use_main_cert=True):
	if use_main_cert or not allow_missing_cert:
		# Get the system certificate info.
		ssl_private_key = os.path.join(os.path.join(env["STORAGE_ROOT"], 'ssl', 'ssl_private_key.pem'))
		ssl_certificate = os.path.join(os.path.join(env["STORAGE_ROOT"], 'ssl', 'ssl_certificate.pem'))
		system_certificate = {
			"private-key": ssl_private_key,
			"certificate": ssl_certificate,
			"primary-domain": env['PRIMARY_HOSTNAME'],
			"certificate_object": load_pem(load_cert_chain(ssl_certificate)[0]),
		}

	if use_main_cert and domain == env['PRIMARY_HOSTNAME']:
		# The primary domain must use the server certificate because
		# it is hard-coded in some service configuration files.
		return system_certificate

	wildcard_domain = re.sub(r"^[^\.]+", "*", domain)
	if domain in ssl_certificates:
		return ssl_certificates[domain]
	elif wildcard_domain in ssl_certificates:
		return ssl_certificates[wildcard_domain]
	elif not allow_missing_cert:
		# No valid certificate is available for this domain! Return default files.
		return system_certificate
	else:
		# No valid certificate is available for this domain.
		return None


# PROVISIONING CERTIFICATES FROM LETSENCRYPT

def get_certificates_to_provision(env, limit_domains=None, show_valid_certs=True):
	# Get a set of domain names that we can provision certificates for
	# using certbot. We start with domains that the box is serving web
	# for and subtract:
	# * domains not in limit_domains if limit_domains is not empty
	# * domains with custom "A" records, i.e. they are hosted elsewhere
	# * domains with actual "A" records that point elsewhere (misconfiguration)
	# * domains that already have certificates that will be valid for a while

	from web_update import get_web_domains
	from status_checks import query_dns, normalize_ip

	existing_certs = get_ssl_certificates(env)

	plausible_web_domains = get_web_domains(env, exclude_dns_elsewhere=False)
	actual_web_domains = get_web_domains(env)

	domains_to_provision = set()
	domains_cant_provision = { }

	for domain in plausible_web_domains:
		# Skip domains that the user doesn't want to provision now.
		if limit_domains and domain not in limit_domains:
			continue

		# Check that there isn't an explicit A/AAAA record.
		if domain not in actual_web_domains:
			domains_cant_provision[domain] = "The domain has a custom DNS A/AAAA record that points the domain elsewhere, so there is no point to installing a TLS certificate here and we could not automatically provision one anyway because provisioning requires access to the website (which isn't here)."

		# Check that the DNS resolves to here.
		else:

			# Does the domain resolve to this machine in public DNS? If not,
			# we can't do domain control validation. For IPv6 is configured,
			# make sure both IPv4 and IPv6 are correct because we don't know
			# how Let's Encrypt will connect.
			bad_dns = []
			for rtype, value in [("A", env["PUBLIC_IP"]), ("AAAA", env.get("PUBLIC_IPV6"))]:
				if not value: continue # IPv6 is not configured
				response = query_dns(domain, rtype)
				if response != normalize_ip(value):
					bad_dns.append(f"{response} ({rtype})")

			if bad_dns:
				domains_cant_provision[domain] = "The domain name does not resolve to this machine: " \
					+ (", ".join(bad_dns)) \
					+ "."

			else:
				# DNS is all good.

				# Check for a good existing cert.
				existing_cert = get_domain_ssl_files(domain, existing_certs, env, use_main_cert=False, allow_missing_cert=True)
				if existing_cert:
					existing_cert_check = check_certificate(domain, existing_cert['certificate'], existing_cert['private-key'],
						warn_if_expiring_soon=14)
					if existing_cert_check[0] == "OK":
						if show_valid_certs:
							domains_cant_provision[domain] = "The domain has a valid certificate already. ({} Certificate: {}, private key {})".format(
								existing_cert_check[1],
								existing_cert['certificate'],
								existing_cert['private-key'])
						continue

				domains_to_provision.add(domain)

	return (domains_to_provision, domains_cant_provision)

def provision_certificates(env, limit_domains):
	# What domains should we provision certificates for? And what
	# errors prevent provisioning for other domains.
	domains, domains_cant_provision = get_certificates_to_provision(env, limit_domains=limit_domains)

	# Build a list of what happened on each domain or domain-set.
	ret = []
	for domain, error in domains_cant_provision.items():
		ret.append({
			"domains": [domain],
			"log": [error],
			"result": "skipped",
		})

	# Break into groups by DNS zone: Group every domain with its parent domain, if
	# its parent domain is in the list of domains to request a certificate for.
	# Start with the zones so that if the zone doesn't need a certificate itself,
	# its children will still be grouped together. Sort the provision domains to
	# put parents ahead of children.
	# Since Let's Encrypt requests are limited to 100 domains at a time,
	# we'll create a list of lists of domains where the inner lists have
	# at most 100 items. By sorting we also get the DNS zone domain as the first
	# entry in each list (unless we overflow beyond 100) which ends up as the
	# primary domain listed in each certificate.
	from dns_update import get_dns_zones
	certs = { }
	for zone, _zonefile in get_dns_zones(env):
		certs[zone] = [[]]
	for domain in sort_domains(domains, env):
		# Does the domain end with any domain we've seen so far.
		for parent in certs:
			if domain.endswith("." + parent):
				# Add this to the parent's list of domains.
				# Start a new group if the list already has
				# 100 items.
				if len(certs[parent][-1]) == 100:
					certs[parent].append([])
				certs[parent][-1].append(domain)
				break
		else:
			# This domain is not a child of any domain we've seen yet, so
			# start a new group. This shouldn't happen since every zone
			# was already added.
			certs[domain] = [[domain]]

	# Flatten to a list of lists of domains (from a mapping). Remove empty
	# lists (zones with no domains that need certs).
	certs = functools.reduce(operator.iadd, certs.values(), [])
	certs = [_ for _ in certs if len(_) > 0]

	# Prepare to provision.

	# Where should we put our Let's Encrypt account info and state cache.
	account_path = os.path.join(env['STORAGE_ROOT'], 'ssl/lets_encrypt')
	if not os.path.exists(account_path):
		os.mkdir(account_path)

	# Provision certificates.
	for domain_list in certs:
		ret.append({
			"domains": domain_list,
			"log": [],
		})
		try:
			# Create a CSR file for our master private key so that certbot
			# uses our private key.
			key_file = os.path.join(env['STORAGE_ROOT'], 'ssl', 'ssl_private_key.pem')
			with tempfile.NamedTemporaryFile() as csr_file:
				# We could use openssl, but certbot requires
				# that the CN domain and SAN domains match
				# the domain list passed to certbot, and adding
				# SAN domains openssl req is ridiculously complicated.
				# subprocess.check_output([
				# 	"openssl", "req", "-new",
				# 	"-key", key_file,
				# 	"-out", csr_file.name,
				# 	"-subj", "/CN=" + domain_list[0],
				# 	"-sha256" ])
				from cryptography import x509
				from cryptography.hazmat.backends import default_backend
				from cryptography.hazmat.primitives.serialization import Encoding
				from cryptography.hazmat.primitives import hashes
				from cryptography.x509.oid import NameOID
				builder = x509.CertificateSigningRequestBuilder()
				builder = builder.subject_name(x509.Name([ x509.NameAttribute(NameOID.COMMON_NAME, domain_list[0]) ]))
				builder = builder.add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
				builder = builder.add_extension(x509.SubjectAlternativeName(
					[x509.DNSName(d) for d in domain_list]
				), critical=False)
				request = builder.sign(load_pem(load_cert_chain(key_file)[0]), hashes.SHA256(), default_backend())
				with open(csr_file.name, "wb") as f:
					f.write(request.public_bytes(Encoding.PEM))

				# Provision, writing to a temporary file.
				webroot = os.path.join(account_path, 'webroot')
				os.makedirs(webroot, exist_ok=True)
				with tempfile.TemporaryDirectory() as d:
					cert_file = os.path.join(d, 'cert_and_chain.pem')
					print("Provisioning TLS certificates for " + ", ".join(domain_list) + ".")
					certbotret = subprocess.check_output([
						"certbot",
						"certonly",
						#"-v", # just enough to see ACME errors
						"--non-interactive", # will fail if user hasn't registered during Mail-in-a-Box setup

						"-d", ",".join(domain_list), # first will be main domain

						"--csr", csr_file.name, # use our private key; unfortunately this doesn't work with auto-renew so we need to save cert manually
						"--cert-path", os.path.join(d, 'cert'), # we only use the full chain
						"--chain-path", os.path.join(d, 'chain'), # we only use the full chain
						"--fullchain-path", cert_file,

						"--webroot", "--webroot-path", webroot,

						"--config-dir", account_path,
						#"--staging",
					], stderr=subprocess.STDOUT).decode("utf8")
					install_cert_copy_file(cert_file, env)

			ret[-1]["log"].append(certbotret)
			ret[-1]["result"] = "installed"
		except subprocess.CalledProcessError as e:
			ret[-1]["log"].append(e.output.decode("utf8"))
			ret[-1]["result"] = "error"
		except Exception as e:
			ret[-1]["log"].append(str(e))
			ret[-1]["result"] = "error"

	# Run post-install steps.
	ret.extend(post_install_func(env))

	# Return what happened with each certificate request.
	return ret

def provision_certificates_cmdline():
	import sys
	from exclusiveprocess import Lock

	from utils import load_environment

	Lock(die=True).forever()
	env = load_environment()

	quiet = False
	domains = []

	for arg in sys.argv[1:]:
		if arg == "-q":
			quiet = True
		else:
			domains.append(arg)

	# Go.
	status = provision_certificates(env, limit_domains=domains)

	# Show what happened.
	for request in status:
		if isinstance(request, str):
			print(request)
		else:
			if quiet and request['result'] == 'skipped':
				continue
			print(request['result'] + ":", ", ".join(request['domains']) + ":")
			for line in request["log"]:
				print(line)
			print()


# INSTALLING A NEW CERTIFICATE FROM THE CONTROL PANEL

def create_csr(domain, ssl_key, country_code, env):
	return shell("check_output", [
				"openssl", "req", "-new",
				"-key", ssl_key,
				"-sha256",
				"-subj", f"/C={country_code}/CN={domain}"])

def install_cert(domain, ssl_cert, ssl_chain, env, raw=False):
	# Write the combined cert+chain to a temporary path and validate that it is OK.
	# The certificate always goes above the chain.
	import tempfile
	fd, fn = tempfile.mkstemp('.pem')
	os.write(fd, (ssl_cert + '\n' + ssl_chain).encode("ascii"))
	os.close(fd)

	# Do validation on the certificate before installing it.
	ssl_private_key = os.path.join(os.path.join(env["STORAGE_ROOT"], 'ssl', 'ssl_private_key.pem'))
	cert_status, cert_status_details = check_certificate(domain, fn, ssl_private_key)
	if cert_status != "OK":
		if cert_status == "SELF-SIGNED":
			cert_status = "This is a self-signed certificate. I can't install that."
		os.unlink(fn)
		if cert_status_details is not None:
			cert_status += " " + cert_status_details
		return cert_status

	# Copy certificate into ssl directory.
	install_cert_copy_file(fn, env)

	# Run post-install steps.
	ret = post_install_func(env)
	if raw: return ret
	return "\n".join(ret)


def install_cert_copy_file(fn, env):
	# Where to put it?
	# Make a unique path for the certificate.
	from cryptography.hazmat.primitives import hashes
	from binascii import hexlify
	cert = load_pem(load_cert_chain(fn)[0])
	_all_domains, cn = get_certificate_domains(cert)
	path = "{}-{}-{}.pem".format(
		safe_domain_name(cn), # common name, which should be filename safe because it is IDNA-encoded, but in case of a malformed cert make sure it's ok to use as a filename
		cert.not_valid_after.date().isoformat().replace("-", ""), # expiration date
		hexlify(cert.fingerprint(hashes.SHA256())).decode("ascii")[0:8], # fingerprint prefix
		)
	ssl_certificate = os.path.join(os.path.join(env["STORAGE_ROOT"], 'ssl', path))

	# Install the certificate.
	os.makedirs(os.path.dirname(ssl_certificate), exist_ok=True)
	shutil.move(fn, ssl_certificate)


def post_install_func(env):
	ret = []

	# Get the certificate to use for PRIMARY_HOSTNAME.
	ssl_certificates = get_ssl_certificates(env)
	cert = get_domain_ssl_files(env['PRIMARY_HOSTNAME'], ssl_certificates, env, use_main_cert=False)
	if not cert:
		# Ruh-row, we don't have any certificate usable
		# for the primary hostname.
		ret.append("there is no valid certificate for " + env['PRIMARY_HOSTNAME'])

	# Symlink the best cert for PRIMARY_HOSTNAME to the system
	# certificate path, which is hard-coded for various purposes, and then
	# restart postfix and dovecot.
	system_ssl_certificate = os.path.join(os.path.join(env["STORAGE_ROOT"], 'ssl', 'ssl_certificate.pem'))
	if cert and os.readlink(system_ssl_certificate) != cert['certificate']:
		# Update symlink.
		ret.append("updating primary certificate")
		ssl_certificate = cert['certificate']
		os.unlink(system_ssl_certificate)
		os.symlink(ssl_certificate, system_ssl_certificate)

		# Restart postfix and dovecot so they pick up the new file.
		shell('check_call', ["/usr/sbin/service", "postfix", "restart"])
		shell('check_call', ["/usr/sbin/service", "dovecot", "restart"])
		ret.append("mail services restarted")

		# The DANE TLSA record will remain valid so long as the private key
		# hasn't changed. We don't ever change the private key automatically.
		# If the user does it, they must manually update DNS.

	# Update the web configuration so nginx picks up the new certificate file.
	from web_update import do_web_update
	ret.append( do_web_update(env) )

	return ret

# VALIDATION OF CERTIFICATES

def check_certificate(domain, ssl_certificate, ssl_private_key, warn_if_expiring_soon=10, rounded_time=False, just_check_domain=False):
	# Check that the ssl_certificate & ssl_private_key files are good
	# for the provided domain.

	from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
	from cryptography.x509 import Certificate

	# The ssl_certificate file may contain a chain of certificates. We'll
	# need to split that up before we can pass anything to openssl or
	# parse them in Python. Parse it with the cryptography library.
	try:
		ssl_cert_chain = load_cert_chain(ssl_certificate)
		cert = load_pem(ssl_cert_chain[0])
		if not isinstance(cert, Certificate): raise ValueError("This is not a certificate file.")
	except ValueError as e:
		return ("There is a problem with the certificate file: %s" % str(e), None)

	# First check that the domain name is one of the names allowed by
	# the certificate.
	if domain is not None:
		certificate_names, _cert_primary_name = get_certificate_domains(cert)

		# Check that the domain appears among the acceptable names, or a wildcard
		# form of the domain name (which is a stricter check than the specs but
		# should work in normal cases).
		wildcard_domain = re.sub(r"^[^\.]+", "*", domain)
		if domain not in certificate_names and wildcard_domain not in certificate_names:
			return ("The certificate is for the wrong domain name. It is for %s."
				% ", ".join(sorted(certificate_names)), None)

	# Second, check that the certificate matches the private key.
	if ssl_private_key is not None:
		try:
			with open(ssl_private_key, 'rb') as f:
				priv_key = load_pem(f.read())
		except ValueError as e:
			return (f"The private key file {ssl_private_key} is not a private key file: {e!s}", None)

		if not isinstance(priv_key, RSAPrivateKey):
			return ("The private key file %s is not a private key file." % ssl_private_key, None)

		if priv_key.public_key().public_numbers() != cert.public_key().public_numbers():
			return ("The certificate does not correspond to the private key at %s." % ssl_private_key, None)

		# We could also use the openssl command line tool to get the modulus
		# listed in each file. The output of each command below looks like "Modulus=XXXXX".
		# $ openssl rsa -inform PEM -noout -modulus -in ssl_private_key
		# $ openssl x509 -in ssl_certificate -noout -modulus

	# Third, check if the certificate is self-signed. Return a special flag string.
	if cert.issuer == cert.subject:
		return ("SELF-SIGNED", None)

	# When selecting which certificate to use for non-primary domains, we check if the primary
	# certificate or a www-parent-domain certificate is good for the domain. There's no need
	# to run extra checks beyond this point.
	if just_check_domain:
		return ("OK", None)

	# Check that the certificate hasn't expired. The datetimes returned by the
	# certificate are 'naive' and in UTC. We need to get the current time in UTC.
	import datetime
	now = datetime.datetime.utcnow()
	if not(cert.not_valid_before <= now <= cert.not_valid_after):
		return (f"The certificate has expired or is not yet valid. It is valid from {cert.not_valid_before} to {cert.not_valid_after}.", None)

	# Next validate that the certificate is valid. This checks whether the certificate
	# is self-signed, that the chain of trust makes sense, that it is signed by a CA
	# that Ubuntu has installed on this machine's list of CAs, and I think that it hasn't
	# expired.

	# The certificate chain has to be passed separately and is given via STDIN.
	# This command returns a non-zero exit status in most cases, so trap errors.
	retcode, verifyoutput = shell('check_output', [
		"openssl",
		"verify", "-verbose",
		"-purpose", "sslserver", "-policy_check",]
		+ ([] if len(ssl_cert_chain) == 1 else ["-untrusted", "/proc/self/fd/0"])
		+ [ssl_certificate],
		input=b"\n\n".join(ssl_cert_chain[1:]),
		trap=True)

	if "self signed" in verifyoutput:
		# Certificate is self-signed. Probably we detected this above.
		return ("SELF-SIGNED", None)

	elif retcode != 0:
		if "unable to get local issuer certificate" in verifyoutput:
			return ("The certificate is missing an intermediate chain or the intermediate chain is incorrect or incomplete. (%s)" % verifyoutput, None)

		# There is some unknown problem. Return the `openssl verify` raw output.
		return ("There is a problem with the certificate.", verifyoutput.strip())

	else:
		# `openssl verify` returned a zero exit status so the cert is currently
		# good.

		# But is it expiring soon?
		cert_expiration_date = cert.not_valid_after
		ndays = (cert_expiration_date-now).days
		if not rounded_time or ndays <= 10:
			# Yikes better renew soon!
			expiry_info = "The certificate expires in %d days on %s." % (ndays, cert_expiration_date.date().isoformat())
		else:
			# We'll renew it with Lets Encrypt.
			expiry_info = "The certificate expires on %s." % cert_expiration_date.date().isoformat()

		if warn_if_expiring_soon and ndays <= warn_if_expiring_soon:
			# Warn on day 10 to give 4 days for us to automatically renew the
			# certificate, which occurs on day 14.
			return ("The certificate is expiring soon: " + expiry_info, None)

		# Return the special OK code.
		return ("OK", expiry_info)

def load_cert_chain(pemfile):
	# A certificate .pem file may contain a chain of certificates.
	# Load the file and split them apart.
	re_pem = rb"(-+BEGIN (?:.+)-+[\r\n]+(?:[A-Za-z0-9+/=]{1,64}[\r\n]+)+-+END (?:.+)-+[\r\n]+)"
	with open(pemfile, "rb") as f:
		pem = f.read() + b"\n" # ensure trailing newline
		pemblocks = re.findall(re_pem, pem)
		if len(pemblocks) == 0:
			msg = "File does not contain valid PEM data."
			raise ValueError(msg)
		return pemblocks

def load_pem(pem):
	# Parse a "---BEGIN .... END---" PEM string and return a Python object for it
	# using classes from the cryptography package.
	from cryptography.x509 import load_pem_x509_certificate
	from cryptography.hazmat.primitives import serialization
	from cryptography.hazmat.backends import default_backend
	pem_type = re.match(b"-+BEGIN (.*?)-+[\r\n]", pem)
	if pem_type is None:
		msg = "File is not a valid PEM-formatted file."
		raise ValueError(msg)
	pem_type = pem_type.group(1)
	if pem_type in {b"RSA PRIVATE KEY", b"PRIVATE KEY"}:
		return serialization.load_pem_private_key(pem, password=None, backend=default_backend())
	if pem_type == b"CERTIFICATE":
		return load_pem_x509_certificate(pem, default_backend())
	raise ValueError("Unsupported PEM object type: " + pem_type.decode("ascii", "replace"))

def get_certificate_domains(cert):
	from cryptography.x509 import DNSName, ExtensionNotFound, OID_COMMON_NAME, OID_SUBJECT_ALTERNATIVE_NAME
	import idna

	names = set()
	cn = None

	# The domain may be found in the Subject Common Name (CN). This comes back as an IDNA (ASCII)
	# string, which is the format we store domains in - so good.
	try:
		cn = cert.subject.get_attributes_for_oid(OID_COMMON_NAME)[0].value
		names.add(cn)
	except IndexError:
		# No common name? Certificate is probably generated incorrectly.
		# But we'll let it error-out when it doesn't find the domain.
		pass

	# ... or be one of the Subject Alternative Names. The cryptography library handily IDNA-decodes
	# the names for us. We must encode back to ASCII, but wildcard certificates can't pass through
	# IDNA encoding/decoding so we must special-case. See https://github.com/pyca/cryptography/pull/2071.
	def idna_decode_dns_name(dns_name):
		if dns_name.startswith("*."):
			return "*." + idna.encode(dns_name[2:]).decode('ascii')
		else:
			return idna.encode(dns_name).decode('ascii')

	try:
		sans = cert.extensions.get_extension_for_oid(OID_SUBJECT_ALTERNATIVE_NAME).value.get_values_for_type(DNSName)
		for san in sans:
			names.add(idna_decode_dns_name(san))
	except ExtensionNotFound:
		pass

	return names, cn

if __name__  == "__main__":
	# Provision certificates.
	provision_certificates_cmdline()



File: management/backup.py

#!/usr/local/lib/mailinabox/env/bin/python

# This script performs a backup of all user data:
# 1) System services are stopped.
# 2) STORAGE_ROOT/backup/before-backup is executed if it exists.
# 3) An incremental encrypted backup is made using duplicity.
# 4) The stopped services are restarted.
# 5) STORAGE_ROOT/backup/after-backup is executed if it exists.

import os, os.path, re, datetime, sys
import dateutil.parser, dateutil.relativedelta, dateutil.tz
import rtyaml
from exclusiveprocess import Lock

from utils import load_environment, shell, wait_for_service

def backup_status(env):
	# If backups are disabled, return no status.
	config = get_backup_config(env)
	if config["target"] == "off":
		return { }

	# Query duplicity to get a list of all full and incremental
	# backups available.

	backups = { }
	now = datetime.datetime.now(dateutil.tz.tzlocal())
	backup_root = os.path.join(env["STORAGE_ROOT"], 'backup')
	backup_cache_dir = os.path.join(backup_root, 'cache')

	def reldate(date, ref, clip):
		if ref < date: return clip
		rd = dateutil.relativedelta.relativedelta(ref, date)
		if rd.years > 1: return "%d years, %d months" % (rd.years, rd.months)
		if rd.years == 1: return "%d year, %d months" % (rd.years, rd.months)
		if rd.months > 1: return "%d months, %d days" % (rd.months, rd.days)
		if rd.months == 1: return "%d month, %d days" % (rd.months, rd.days)
		if rd.days >= 7: return "%d days" % rd.days
		if rd.days > 1: return "%d days, %d hours" % (rd.days, rd.hours)
		if rd.days == 1: return "%d day, %d hours" % (rd.days, rd.hours)
		return "%d hours, %d minutes" % (rd.hours, rd.minutes)

	# Get duplicity collection status and parse for a list of backups.
	def parse_line(line):
		keys = line.strip().split()
		date = dateutil.parser.parse(keys[1]).astimezone(dateutil.tz.tzlocal())
		return {
			"date": keys[1],
			"date_str": date.strftime("%Y-%m-%d %X") + " " + now.tzname(),
			"date_delta": reldate(date, now, "the future?"),
			"full": keys[0] == "full",
			"size": 0, # collection-status doesn't give us the size
			"volumes": int(keys[2]), # number of archive volumes for this backup (not really helpful)
		}

	code, collection_status = shell('check_output', [
		"/usr/bin/duplicity",
		"collection-status",
		"--archive-dir", backup_cache_dir,
		"--gpg-options", "'--cipher-algo=AES256'",
		"--log-fd", "1",
		*get_duplicity_additional_args(env),
		get_duplicity_target_url(config)
		],
		get_duplicity_env_vars(env),
		trap=True)
	if code != 0:
		# Command failed. This is likely due to an improperly configured remote
		# destination for the backups or the last backup job terminated unexpectedly.
		raise Exception("Something is wrong with the backup: " + collection_status)
	for line in collection_status.split('\n'):
		if line.startswith((" full", " inc")):
			backup = parse_line(line)
			backups[backup["date"]] = backup

	# Look at the target directly to get the sizes of each of the backups. There is more than one file per backup.
	# Starting with duplicity in Ubuntu 18.04, "signatures" files have dates in their
	# filenames that are a few seconds off the backup date and so don't line up
	# with the list of backups we have. Track unmatched files so we know how much other
	# space is used for those.
	unmatched_file_size = 0
	for fn, size in list_target_files(config):
		m = re.match(r"duplicity-(full|full-signatures|(inc|new-signatures)\.(?P<incbase>\d+T\d+Z)\.to)\.(?P<date>\d+T\d+Z)\.", fn)
		if not m: continue # not a part of a current backup chain
		key = m.group("date")
		if key in backups:
			backups[key]["size"] += size
		else:
			unmatched_file_size += size

	# Ensure the rows are sorted reverse chronologically.
	# This is relied on by should_force_full() and the next step.
	backups = sorted(backups.values(), key = lambda b : b["date"], reverse=True)

	# Get the average size of incremental backups, the size of the
	# most recent full backup, and the date of the most recent
	# backup and the most recent full backup.
	incremental_count = 0
	incremental_size = 0
	first_date = None
	first_full_size = None
	first_full_date = None
	for bak in backups:
		if first_date is None:
			first_date = dateutil.parser.parse(bak["date"])
		if bak["full"]:
			first_full_size = bak["size"]
			first_full_date = dateutil.parser.parse(bak["date"])
			break
		incremental_count += 1
		incremental_size += bak["size"]

	# When will the most recent backup be deleted? It won't be deleted if the next
	# backup is incremental, because the increments rely on all past increments.
	# So first guess how many more incremental backups will occur until the next
	# full backup. That full backup frees up this one to be deleted. But, the backup
	# must also be at least min_age_in_days old too.
	deleted_in = None
	if incremental_count > 0 and incremental_size > 0 and first_full_size is not None:
		# How many days until the next incremental backup? First, the part of
		# the algorithm based on increment sizes:
		est_days_to_next_full = (.5 * first_full_size - incremental_size) / (incremental_size/incremental_count)
		est_time_of_next_full = first_date + datetime.timedelta(days=est_days_to_next_full)

		# ...And then the part of the algorithm based on full backup age:
		est_time_of_next_full = min(est_time_of_next_full, first_full_date + datetime.timedelta(days=config["min_age_in_days"]*10+1))

		# It still can't be deleted until it's old enough.
		est_deleted_on = max(est_time_of_next_full, first_date + datetime.timedelta(days=config["min_age_in_days"]))

		deleted_in = "approx. %d days" % round((est_deleted_on-now).total_seconds()/60/60/24 + .5)

	# When will a backup be deleted? Set the deleted_in field of each backup.
	saw_full = False
	for bak in backups:
		if deleted_in:
			# The most recent increment in a chain and all of the previous backups
			# it relies on are deleted at the same time.
			bak["deleted_in"] = deleted_in
		if bak["full"]:
			# Reset when we get to a full backup. A new chain start *next*.
			saw_full = True
			deleted_in = None
		elif saw_full and not deleted_in:
			# We're now on backups prior to the most recent full backup. These are
			# free to be deleted as soon as they are min_age_in_days old.
			deleted_in = reldate(now, dateutil.parser.parse(bak["date"]) + datetime.timedelta(days=config["min_age_in_days"]), "on next daily backup")
			bak["deleted_in"] = deleted_in

	return {
		"backups": backups,
		"unmatched_file_size": unmatched_file_size,
	}

def should_force_full(config, env):
	# Force a full backup when the total size of the increments
	# since the last full backup is greater than half the size
	# of that full backup.
	inc_size = 0
	for bak in backup_status(env)["backups"]:
		if not bak["full"]:
			# Scan through the incremental backups cumulating
			# size...
			inc_size += bak["size"]
		else:
			# ...until we reach the most recent full backup.
			# Return if we should to a full backup, which is based
			# on the size of the increments relative to the full
			# backup, as well as the age of the full backup.
			if inc_size > .5*bak["size"]:
				return True
			if dateutil.parser.parse(bak["date"]) + datetime.timedelta(days=config["min_age_in_days"]*10+1) < datetime.datetime.now(dateutil.tz.tzlocal()):
				return True
			return False
	else:
		# If we got here there are no (full) backups, so make one.
		# (I love for/else blocks. Here it's just to show off.)
		return True

def get_passphrase(env):
	# Get the encryption passphrase. secret_key.txt is 2048 random
	# bits base64-encoded and with line breaks every 65 characters.
	# gpg will only take the first line of text, so sanity check that
	# that line is long enough to be a reasonable passphrase. It
	# only needs to be 43 base64-characters to match AES256's key
	# length of 32 bytes.
	backup_root = os.path.join(env["STORAGE_ROOT"], 'backup')
	with open(os.path.join(backup_root, 'secret_key.txt'), encoding="utf-8") as f:
		passphrase = f.readline().strip()
	if len(passphrase) < 43: raise Exception("secret_key.txt's first line is too short!")

	return passphrase

def get_duplicity_target_url(config):
	target = config["target"]

	if get_target_type(config) == "s3":
		from urllib.parse import urlsplit, urlunsplit
		target = list(urlsplit(target))

		# Although we store the S3 hostname in the target URL,
		# duplicity no longer accepts it in the target URL. The hostname in
		# the target URL must be the bucket name. The hostname is passed
		# via get_duplicity_additional_args. Move the first part of the
		# path (the bucket name) into the hostname URL component, and leave
		# the rest for the path. (The S3 region name is also stored in the
		# hostname part of the URL, in the username portion, which we also
		# have to drop here).
		target[1], target[2] = target[2].lstrip('/').split('/', 1)

		target = urlunsplit(target)

	return target

def get_duplicity_additional_args(env):
	config = get_backup_config(env)

	if get_target_type(config) == 'rsync':
		# Extract a port number for the ssh transport.  Duplicity accepts the
		# optional port number syntax in the target, but it doesn't appear to act
		# on it, so we set the ssh port explicitly via the duplicity options.
		from urllib.parse import urlsplit
		try:
			port = urlsplit(config["target"]).port
		except ValueError:
			port = 22
		if port is None:
			port = 22

		return [
			f"--ssh-options='-i /root/.ssh/id_rsa_miab -p {port}'",
			f"--rsync-options='-e \"/usr/bin/ssh -oStrictHostKeyChecking=no -oBatchMode=yes -p {port} -i /root/.ssh/id_rsa_miab\"'",
		]
	elif get_target_type(config) == 's3':
		# See note about hostname in get_duplicity_target_url.
		# The region name, which is required by some non-AWS endpoints,
		# is saved inside the username portion of the URL.
		from urllib.parse import urlsplit, urlunsplit
		target = urlsplit(config["target"])
		endpoint_url = urlunsplit(("https", target.hostname, '', '', ''))
		args = ["--s3-endpoint-url", endpoint_url]
		if target.username: # region name is stuffed here
			args += ["--s3-region-name", target.username]
		return args

	return []

def get_duplicity_env_vars(env):
	config = get_backup_config(env)

	env = { "PASSPHRASE" : get_passphrase(env) }

	if get_target_type(config) == 's3':
		env["AWS_ACCESS_KEY_ID"] = config["target_user"]
		env["AWS_SECRET_ACCESS_KEY"] = config["target_pass"]

	return env

def get_target_type(config):
	return config["target"].split(":")[0]

def perform_backup(full_backup):
	env = load_environment()

	# Create an global exclusive lock so that the backup script
	# cannot be run more than one.
	Lock(die=True).forever()

	config = get_backup_config(env)
	backup_root = os.path.join(env["STORAGE_ROOT"], 'backup')
	backup_cache_dir = os.path.join(backup_root, 'cache')
	backup_dir = os.path.join(backup_root, 'encrypted')

	# Are backups disabled?
	if config["target"] == "off":
		return

	# On the first run, always do a full backup. Incremental
	# will fail. Otherwise do a full backup when the size of
	# the increments since the most recent full backup are
	# large.
	try:
		full_backup = full_backup or should_force_full(config, env)
	except Exception as e:
		# This was the first call to duplicity, and there might
		# be an error already.
		print(e)
		sys.exit(1)

	# Stop services.
	def service_command(service, command, quit=None):
		# Execute silently, but if there is an error then display the output & exit.
		code, ret = shell('check_output', ["/usr/sbin/service", service, command], capture_stderr=True, trap=True)
		if code != 0:
			print(ret)
			if quit:
				sys.exit(code)

	service_command("php8.0-fpm", "stop", quit=True)
	service_command("postfix", "stop", quit=True)
	service_command("dovecot", "stop", quit=True)
	service_command("postgrey", "stop", quit=True)

	# Execute a pre-backup script that copies files outside the homedir.
	# Run as the STORAGE_USER user, not as root. Pass our settings in
	# environment variables so the script has access to STORAGE_ROOT.
	pre_script = os.path.join(backup_root, 'before-backup')
	if os.path.exists(pre_script):
		shell('check_call',
			['su', env['STORAGE_USER'], '-c', pre_script, config["target"]],
			env=env)

	# Run a backup of STORAGE_ROOT (but excluding the backups themselves!).
	# --allow-source-mismatch is needed in case the box's hostname is changed
	# after the first backup. See #396.
	try:
		shell('check_call', [
			"/usr/bin/duplicity",
			"full" if full_backup else "incr",
			"--verbosity", "warning", "--no-print-statistics",
			"--archive-dir", backup_cache_dir,
			"--exclude", backup_root,
			"--volsize", "250",
			"--gpg-options", "'--cipher-algo=AES256'",
			"--allow-source-mismatch",
			*get_duplicity_additional_args(env),
			env["STORAGE_ROOT"],
			get_duplicity_target_url(config),
			],
			get_duplicity_env_vars(env))
	finally:
		# Start services again.
		service_command("postgrey", "start", quit=False)
		service_command("dovecot", "start", quit=False)
		service_command("postfix", "start", quit=False)
		service_command("php8.0-fpm", "start", quit=False)

	# Remove old backups. This deletes all backup data no longer needed
	# from more than 3 days ago.
	shell('check_call', [
		"/usr/bin/duplicity",
		"remove-older-than",
		"%dD" % config["min_age_in_days"],
		"--verbosity", "error",
		"--archive-dir", backup_cache_dir,
		"--force",
		*get_duplicity_additional_args(env),
		get_duplicity_target_url(config)
		],
		get_duplicity_env_vars(env))

	# From duplicity's manual:
	# "This should only be necessary after a duplicity session fails or is
	# aborted prematurely."
	# That may be unlikely here but we may as well ensure we tidy up if
	# that does happen - it might just have been a poorly timed reboot.
	shell('check_call', [
		"/usr/bin/duplicity",
		"cleanup",
		"--verbosity", "error",
		"--archive-dir", backup_cache_dir,
		"--force",
		*get_duplicity_additional_args(env),
		get_duplicity_target_url(config)
		],
		get_duplicity_env_vars(env))

	# Change ownership of backups to the user-data user, so that the after-bcakup
	# script can access them.
	if get_target_type(config) == 'file':
		shell('check_call', ["/bin/chown", "-R", env["STORAGE_USER"], backup_dir])

	# Execute a post-backup script that does the copying to a remote server.
	# Run as the STORAGE_USER user, not as root. Pass our settings in
	# environment variables so the script has access to STORAGE_ROOT.
	post_script = os.path.join(backup_root, 'after-backup')
	if os.path.exists(post_script):
		shell('check_call',
			['su', env['STORAGE_USER'], '-c', post_script, config["target"]],
			env=env)

	# Our nightly cron job executes system status checks immediately after this
	# backup. Since it checks that dovecot and postfix are running, block for a
	# bit (maximum of 10 seconds each) to give each a chance to finish restarting
	# before the status checks might catch them down. See #381.
	wait_for_service(25, True, env, 10)
	wait_for_service(993, True, env, 10)

def run_duplicity_verification():
	env = load_environment()
	backup_root = os.path.join(env["STORAGE_ROOT"], 'backup')
	config = get_backup_config(env)
	backup_cache_dir = os.path.join(backup_root, 'cache')

	shell('check_call', [
		"/usr/bin/duplicity",
		"--verbosity", "info",
		"verify",
		"--compare-data",
		"--archive-dir", backup_cache_dir,
		"--exclude", backup_root,
		*get_duplicity_additional_args(env),
		get_duplicity_target_url(config),
		env["STORAGE_ROOT"],
	], get_duplicity_env_vars(env))

def run_duplicity_restore(args):
	env = load_environment()
	config = get_backup_config(env)
	backup_cache_dir = os.path.join(env["STORAGE_ROOT"], 'backup', 'cache')
	shell('check_call', [
		"/usr/bin/duplicity",
		"restore",
		"--archive-dir", backup_cache_dir,
		*get_duplicity_additional_args(env),
		get_duplicity_target_url(config),
		*args],
		get_duplicity_env_vars(env))

def print_duplicity_command():
	import shlex
	env = load_environment()
	config = get_backup_config(env)
	backup_cache_dir = os.path.join(env["STORAGE_ROOT"], 'backup', 'cache')
	for k, v in get_duplicity_env_vars(env).items():
		print(f"export {k}={shlex.quote(v)}")
	print("duplicity", "{command}", shlex.join([
		"--archive-dir", backup_cache_dir,
		*get_duplicity_additional_args(env),
		get_duplicity_target_url(config)
		]))

def list_target_files(config):
	import urllib.parse
	try:
		target = urllib.parse.urlparse(config["target"])
	except ValueError:
		return "invalid target"

	if target.scheme == "file":
		return [(fn, os.path.getsize(os.path.join(target.path, fn))) for fn in os.listdir(target.path)]

	elif target.scheme == "rsync":
		rsync_fn_size_re = re.compile(r'.*    ([^ ]*) [^ ]* [^ ]* (.*)')
		rsync_target = '{host}:{path}'

		# Strip off any trailing port specifier because it's not valid in rsync's
		# DEST syntax.  Explicitly set the port number for the ssh transport.
		user_host, *_ = target.netloc.rsplit(':', 1)
		try:
			port = target.port
		except ValueError:
			 port = 22
		if port is None:
			port = 22

		target_path = target.path
		if not target_path.endswith('/'):
			target_path = target_path + '/'
		if target_path.startswith('/'):
			target_path = target_path[1:]

		rsync_command = [ 'rsync',
					'-e',
					f'/usr/bin/ssh -i /root/.ssh/id_rsa_miab -oStrictHostKeyChecking=no -oBatchMode=yes -p {port}',
					'--list-only',
					'-r',
					rsync_target.format(
						host=user_host,
						path=target_path)
				]

		code, listing = shell('check_output', rsync_command, trap=True, capture_stderr=True)
		if code == 0:
			ret = []
			for l in listing.split('\n'):
				match = rsync_fn_size_re.match(l)
				if match:
					ret.append( (match.groups()[1], int(match.groups()[0].replace(',',''))) )
			return ret
		else:
			if 'Permission denied (publickey).' in listing:
				reason = "Invalid user or check you correctly copied the SSH key."
			elif 'No such file or directory' in listing:
				reason = f"Provided path {target_path} is invalid."
			elif 'Network is unreachable' in listing:
				reason = f"The IP address {target.hostname} is unreachable."
			elif 'Could not resolve hostname' in listing:
				reason = f"The hostname {target.hostname} cannot be resolved."
			else:
				reason = ("Unknown error."
						"Please check running 'management/backup.py --verify'"
						"from mailinabox sources to debug the issue.")
			msg = f"Connection to rsync host failed: {reason}"
			raise ValueError(msg)

	elif target.scheme == "s3":
		import boto3.s3
		from botocore.exceptions import ClientError

		# separate bucket from path in target
		bucket = target.path[1:].split('/')[0]
		path = '/'.join(target.path[1:].split('/')[1:]) + '/'

		# If no prefix is specified, set the path to '', otherwise boto won't list the files
		if path == '/':
			path = ''

		if bucket == "":
			msg = "Enter an S3 bucket name."
			raise ValueError(msg)

		# connect to the region & bucket
		try:
			s3 = boto3.client('s3', \
				endpoint_url=f'https://{target.hostname}', \
				aws_access_key_id=config['target_user'], \
				aws_secret_access_key=config['target_pass'])
			bucket_objects = s3.list_objects_v2(Bucket=bucket, Prefix=path)['Contents']
			backup_list = [(key['Key'][len(path):], key['Size']) for key in bucket_objects]
		except ClientError as e:
			raise ValueError(e)
		return backup_list
	elif target.scheme == 'b2':
		from b2sdk.v1 import InMemoryAccountInfo, B2Api
		from b2sdk.v1.exception import NonExistentBucket
		info = InMemoryAccountInfo()
		b2_api = B2Api(info)

		# Extract information from target
		b2_application_keyid = target.netloc[:target.netloc.index(':')]
		b2_application_key = urllib.parse.unquote(target.netloc[target.netloc.index(':')+1:target.netloc.index('@')])
		b2_bucket = target.netloc[target.netloc.index('@')+1:]

		try:
			b2_api.authorize_account("production", b2_application_keyid, b2_application_key)
			bucket = b2_api.get_bucket_by_name(b2_bucket)
		except NonExistentBucket:
			msg = "B2 Bucket does not exist. Please double check your information!"
			raise ValueError(msg)
		return [(key.file_name, key.size) for key, _ in bucket.ls()]

	else:
		raise ValueError(config["target"])


def backup_set_custom(env, target, target_user, target_pass, min_age):
	config = get_backup_config(env, for_save=True)

	# min_age must be an int
	if isinstance(min_age, str):
		min_age = int(min_age)

	config["target"] = target
	config["target_user"] = target_user
	config["target_pass"] = target_pass
	config["min_age_in_days"] = min_age

	# Validate.
	try:
		if config["target"] not in {"off", "local"}:
			# these aren't supported by the following function, which expects a full url in the target key,
			# which is what is there except when loading the config prior to saving
			list_target_files(config)
	except ValueError as e:
		return str(e)

	write_backup_config(env, config)

	return "OK"

def get_backup_config(env, for_save=False, for_ui=False):
	backup_root = os.path.join(env["STORAGE_ROOT"], 'backup')

	# Defaults.
	config = {
		"min_age_in_days": 3,
		"target": "local",
	}

	# Merge in anything written to custom.yaml.
	try:
		with open(os.path.join(backup_root, 'custom.yaml'), encoding="utf-8") as f:
			custom_config = rtyaml.load(f)
		if not isinstance(custom_config, dict): raise ValueError # caught below
		config.update(custom_config)
	except:
		pass

	# When updating config.yaml, don't do any further processing on what we find.
	if for_save:
		return config

	# When passing this back to the admin to show the current settings, do not include
	# authentication details. The user will have to re-enter it.
	if for_ui:
		for field in ("target_user", "target_pass"):
			if field in config:
				del config[field]

	# helper fields for the admin
	config["file_target_directory"] = os.path.join(backup_root, 'encrypted')
	config["enc_pw_file"] = os.path.join(backup_root, 'secret_key.txt')
	if config["target"] == "local":
		# Expand to the full URL.
		config["target"] = "file://" + config["file_target_directory"]
	ssh_pub_key = os.path.join('/root', '.ssh', 'id_rsa_miab.pub')
	if os.path.exists(ssh_pub_key):
		with open(ssh_pub_key, encoding="utf-8") as f:
			config["ssh_pub_key"] = f.read()

	return config

def write_backup_config(env, newconfig):
	backup_root = os.path.join(env["STORAGE_ROOT"], 'backup')
	with open(os.path.join(backup_root, 'custom.yaml'), "w", encoding="utf-8") as f:
		f.write(rtyaml.dump(newconfig))

if __name__ == "__main__":
	if sys.argv[-1] == "--verify":
		# Run duplicity's verification command to check a) the backup files
		# are readable, and b) report if they are up to date.
		run_duplicity_verification()

	elif sys.argv[-1] == "--list":
		# List the saved backup files.
		for fn, size in list_target_files(get_backup_config(load_environment())):
			print(f"{fn}\t{size}")

	elif sys.argv[-1] == "--status":
		# Show backup status.
		ret = backup_status(load_environment())
		print(rtyaml.dump(ret["backups"]))
		print("Storage for unmatched files:", ret["unmatched_file_size"])

	elif len(sys.argv) >= 2 and sys.argv[1] == "--restore":
		# Run duplicity restore. Rest of command line passed as arguments
		# to duplicity. The restore path should be specified.
		run_duplicity_restore(sys.argv[2:])

	elif sys.argv[-1] == "--duplicity-command":
		print_duplicity_command()

	else:
		# Perform a backup. Add --full to force a full backup rather than
		# possibly performing an incremental backup.
		full_backup = "--full" in sys.argv
		perform_backup(full_backup)



File: management/status_checks.py

#!/usr/local/lib/mailinabox/env/bin/python
#
# Checks that the upstream DNS has been set correctly and that
# TLS certificates have been signed, etc., and if not tells the user
# what to do next.

import sys, os, os.path, re, datetime, multiprocessing.pool
import asyncio

import dns.reversename, dns.resolver
import idna
import psutil
import postfix_mta_sts_resolver.resolver

from dns_update import get_dns_zones, build_tlsa_record, get_custom_dns_config, get_secondary_dns, get_custom_dns_records
from web_update import get_web_domains, get_domains_with_a_records
from ssl_certificates import get_ssl_certificates, get_domain_ssl_files, check_certificate
from mailconfig import get_mail_domains, get_mail_aliases

from utils import shell, sort_domains, load_env_vars_from_file, load_settings, get_ssh_port, get_ssh_config_value

def get_services():
	return [
		{ "name": "Local DNS (bind9)", "port": 53, "public": False, },
		#{ "name": "NSD Control", "port": 8952, "public": False, },
		{ "name": "Local DNS Control (bind9/rndc)", "port": 953, "public": False, },
		{ "name": "Dovecot LMTP LDA", "port": 10026, "public": False, },
		{ "name": "Postgrey", "port": 10023, "public": False, },
		{ "name": "Spamassassin", "port": 10025, "public": False, },
		{ "name": "OpenDKIM", "port": 8891, "public": False, },
		{ "name": "OpenDMARC", "port": 8893, "public": False, },
		{ "name": "Mail-in-a-Box Management Daemon", "port": 10222, "public": False, },
		{ "name": "SSH Login (ssh)", "port": get_ssh_port(), "public": True, },
		{ "name": "Public DNS (nsd4)", "port": 53, "public": True, },
		{ "name": "Incoming Mail (SMTP/postfix)", "port": 25, "public": True, },
		{ "name": "Outgoing Mail (SMTP 465/postfix)", "port": 465, "public": True, },
		{ "name": "Outgoing Mail (SMTP 587/postfix)", "port": 587, "public": True, },
		#{ "name": "Postfix/master", "port": 10587, "public": True, },
		{ "name": "IMAPS (dovecot)", "port": 993, "public": True, },
		{ "name": "Mail Filters (Sieve/dovecot)", "port": 4190, "public": True, },
		{ "name": "HTTP Web (nginx)", "port": 80, "public": True, },
		{ "name": "HTTPS Web (nginx)", "port": 443, "public": True, },
	]

def run_checks(rounded_values, env, output, pool, domains_to_check=None):
	# run systems checks
	output.add_heading("System")

	# check that services are running
	if not run_services_checks(env, output, pool):
		# If critical services are not running, stop. If bind9 isn't running,
		# all later DNS checks will timeout and that will take forever to
		# go through, and if running over the web will cause a fastcgi timeout.
		return

	# clear bind9's DNS cache so our DNS checks are up to date
	# (ignore errors; if bind9/rndc isn't running we'd already report
	# that in run_services checks.)
	shell('check_call', ["/usr/sbin/rndc", "flush"], trap=True)

	run_system_checks(rounded_values, env, output)

	# perform other checks asynchronously

	run_network_checks(env, output)
	run_domain_checks(rounded_values, env, output, pool, domains_to_check=domains_to_check)

def run_services_checks(env, output, pool):
	# Check that system services are running.
	all_running = True
	fatal = False
	ret = pool.starmap(check_service, ((i, service, env) for i, service in enumerate(get_services())), chunksize=1)
	for _i, running, fatal2, output2 in sorted(ret):
		if output2 is None: continue # skip check (e.g. no port was set, e.g. no sshd)
		all_running = all_running and running
		fatal = fatal or fatal2
		output2.playback(output)

	# Check fail2ban.
	code, ret = shell('check_output', ["fail2ban-client", "status"], capture_stderr=True, trap=True)
	if code != 0:
		output.print_error("fail2ban is not running.")
		all_running = False

	if all_running:
		output.print_ok("All system services are running.")

	return not fatal

def check_service(i, service, env):
	if not service["port"]:
		# Skip check (no port, e.g. no sshd).
		return (i, None, None, None)

	output = BufferedOutput()
	running = False
	fatal = False

	# Helper function to make a connection to the service, since we try
	# up to three ways (localhost, IPv4 address, IPv6 address).
	def try_connect(ip):
		# Connect to the given IP address on the service's port with a one-second timeout.
		import socket
		s = socket.socket(socket.AF_INET if ":" not in ip else socket.AF_INET6, socket.SOCK_STREAM)
		s.settimeout(1)
		try:
			s.connect((ip, service["port"]))
			return True
		except OSError:
			# timed out or some other odd error
			return False
		finally:
			s.close()

	if service["public"]:
		# Service should be publicly accessible.
		if try_connect(env["PUBLIC_IP"]):
			# IPv4 ok.
			if not env.get("PUBLIC_IPV6") or service.get("ipv6") is False or try_connect(env["PUBLIC_IPV6"]):
				# No IPv6, or service isn't meant to run on IPv6, or IPv6 is good.
				running = True

			# IPv4 ok but IPv6 failed. Try the PRIVATE_IPV6 address to see if the service is bound to the interface.
			elif service["port"] != 53 and try_connect(env["PRIVATE_IPV6"]):
				output.print_error("%s is running (and available over IPv4 and the local IPv6 address), but it is not publicly accessible at %s:%d." % (service['name'], env['PUBLIC_IPV6'], service['port']))
			else:
				output.print_error("%s is running and available over IPv4 but is not accessible over IPv6 at %s port %d." % (service['name'], env['PUBLIC_IPV6'], service['port']))

		# IPv4 failed. Try the private IP to see if the service is running but not accessible (except DNS because a different service runs on the private IP).
		elif service["port"] != 53 and try_connect("127.0.0.1"):
			output.print_error("%s is running but is not publicly accessible at %s:%d." % (service['name'], env['PUBLIC_IP'], service['port']))
		else:
			output.print_error("%s is not running (port %d)." % (service['name'], service['port']))

		# Why is nginx not running?
		if not running and service["port"] in {80, 443}:
			output.print_line(shell('check_output', ['nginx', '-t'], capture_stderr=True, trap=True)[1].strip())

	# Service should be running locally.
	elif try_connect("127.0.0.1"):
		running = True
	else:
		output.print_error("%s is not running (port %d)." % (service['name'], service['port']))

	# Flag if local DNS is not running.
	if not running and service["port"] == 53 and service["public"] is False:
		fatal = True

	return (i, running, fatal, output)

def run_system_checks(rounded_values, env, output):
	check_ssh_password(env, output)
	check_software_updates(env, output)
	check_miab_version(env, output)
	check_system_aliases(env, output)
	check_free_disk_space(rounded_values, env, output)
	check_free_memory(rounded_values, env, output)

def check_ufw(env, output):
	if not os.path.isfile('/usr/sbin/ufw'):
		output.print_warning("""The ufw program was not installed. If your system is able to run iptables, rerun the setup.""")
		return

	code, ufw = shell('check_output', ['ufw', 'status'], trap=True)

	if code != 0:
		# The command failed, it's safe to say the firewall is disabled
		output.print_warning("""The firewall is not working on this machine. An error was received
					while trying to check the firewall. To investigate run 'sudo ufw status'.""")
		return

	ufw = ufw.splitlines()
	if ufw[0] == "Status: active":
		not_allowed_ports = 0
		for service in get_services():
			if service["public"] and not is_port_allowed(ufw, service["port"]):
				not_allowed_ports += 1
				output.print_error("Port {} ({}) should be allowed in the firewall, please re-run the setup.".format(service["port"], service["name"]))

		if not_allowed_ports == 0:
			output.print_ok("Firewall is active.")
	else:
		output.print_warning("""The firewall is disabled on this machine. This might be because the system
			is protected by an external firewall. We can't protect the system against bruteforce attacks
			without the local firewall active. Connect to the system via ssh and try to run: ufw enable.""")

def is_port_allowed(ufw, port):
	return any(re.match(str(port) +"[/ \t].*", item) for item in ufw)

def check_ssh_password(env, output):
	config_value = get_ssh_config_value("passwordauthentication")
	if config_value:
		if config_value == "no":
			output.print_ok("SSH disallows password-based login.")
		else:
			output.print_error("""The SSH server on this machine permits password-based login. A more secure
				way to log in is using a public key. Add your SSH public key to $HOME/.ssh/authorized_keys, check
				that you can log in without a password, set the option 'PasswordAuthentication no' in
				/etc/ssh/sshd_config, and then restart the openssh via 'sudo service ssh restart'.""")

def is_reboot_needed_due_to_package_installation():
	return os.path.exists("/var/run/reboot-required")

def check_software_updates(env, output):
	# Check for any software package updates.
	pkgs = list_apt_updates(apt_update=False)
	if is_reboot_needed_due_to_package_installation():
		output.print_error("System updates have been installed and a reboot of the machine is required.")
	elif len(pkgs) == 0:
		output.print_ok("System software is up to date.")
	else:
		output.print_error("There are %d software packages that can be updated." % len(pkgs))
		for p in pkgs:
			output.print_line("{} ({})".format(p["package"], p["version"]))

def check_system_aliases(env, output):
	# Check that the administrator alias exists since that's where all
	# admin email is automatically directed.
	check_alias_exists("System administrator address", "administrator@" + env['PRIMARY_HOSTNAME'], env, output)

def check_free_disk_space(rounded_values, env, output):
	# Check free disk space.
	st = os.statvfs(env['STORAGE_ROOT'])
	bytes_total = st.f_blocks * st.f_frsize
	bytes_free = st.f_bavail * st.f_frsize
	disk_msg = "The disk has %.2f GB space remaining." % (bytes_free/1024.0/1024.0/1024.0)
	if bytes_free > .3 * bytes_total:
		if rounded_values: disk_msg = "The disk has more than 30% free space."
		output.print_ok(disk_msg)
	elif bytes_free > .15 * bytes_total:
		if rounded_values: disk_msg = "The disk has less than 30% free space."
		output.print_warning(disk_msg)
	else:
		if rounded_values: disk_msg = "The disk has less than 15% free space."
		output.print_error(disk_msg)

	# Check that there's only one duplicity cache. If there's more than one,
	# it's probably no longer in use, and we can recommend clearing the cache
	# to save space. The cache directory may not exist yet, which is OK.
	backup_cache_path = os.path.join(env['STORAGE_ROOT'], 'backup/cache')
	try:
		backup_cache_count = len(os.listdir(backup_cache_path))
	except:
		backup_cache_count = 0
	if backup_cache_count > 1:
		output.print_warning(f"The backup cache directory {backup_cache_path} has more than one backup target cache. Consider clearing this directory to save disk space.")

def check_free_memory(rounded_values, env, output):
	# Check free memory.
	percent_free = 100 - psutil.virtual_memory().percent
	memory_msg = "System memory is %s%% free." % str(round(percent_free))
	if percent_free >= 20:
		if rounded_values: memory_msg = "System free memory is at least 20%."
		output.print_ok(memory_msg)
	elif percent_free >= 10:
		if rounded_values: memory_msg = "System free memory is below 20%."
		output.print_warning(memory_msg)
	else:
		if rounded_values: memory_msg = "System free memory is below 10%."
		output.print_error(memory_msg)

def run_network_checks(env, output):
	# Also see setup/network-checks.sh.

	output.add_heading("Network")

	check_ufw(env, output)

	# Stop if we cannot make an outbound connection on port 25. Many residential
	# networks block outbound port 25 to prevent their network from sending spam.
	# See if we can reach one of Google's MTAs with a 5-second timeout.
	_code, ret = shell("check_call", ["/bin/nc", "-z", "-w5", "aspmx.l.google.com", "25"], trap=True)
	if ret == 0:
		output.print_ok("Outbound mail (SMTP port 25) is not blocked.")
	else:
		output.print_error("""Outbound mail (SMTP port 25) seems to be blocked by your network. You
			will not be able to send any mail. Many residential networks block port 25 to prevent hijacked
			machines from being able to send spam. A quick connection test to Google's mail server on port 25
			failed.""")

	# Stop if the IPv4 address is listed in the ZEN Spamhaus Block List.
	# The user might have ended up on an IP address that was previously in use
	# by a spammer, or the user may be deploying on a residential network. We
	# will not be able to reliably send mail in these cases.
	
	# See https://www.spamhaus.org/news/article/807/using-our-public-mirrors-check-your-return-codes-now. for
	# information on spamhaus return codes
	rev_ip4 = ".".join(reversed(env['PUBLIC_IP'].split('.')))
	zen = query_dns(rev_ip4+'.zen.spamhaus.org', 'A', nxdomain=None)
	if zen is None:
		output.print_ok("IP address is not blacklisted by zen.spamhaus.org.")
	elif zen == "[timeout]":
		output.print_warning("Connection to zen.spamhaus.org timed out. Could not determine whether this box's IP address is blacklisted. Please try again later.")
	elif zen == "[Not Set]":
		output.print_warning("Could not connect to zen.spamhaus.org. Could not determine whether this box's IP address is blacklisted. Please try again later.")
	elif zen == "127.255.255.252":
		output.print_warning("Incorrect spamhaus query: %s. Could not determine whether this box's IP address is blacklisted." % (rev_ip4+'.zen.spamhaus.org'))
	elif zen == "127.255.255.254":
		output.print_warning("Mail-in-a-Box is configured to use a public DNS server. This is not supported by spamhaus. Could not determine whether this box's IP address is blacklisted.")
	elif zen == "127.255.255.255":
		output.print_warning("Too many queries have been performed on the spamhaus server. Could not determine whether this box's IP address is blacklisted.")
	else:
		output.print_error("""The IP address of this machine {} is listed in the Spamhaus Block List (code {}),
			which may prevent recipients from receiving your email. See http://www.spamhaus.org/query/ip/{}.""".format(env['PUBLIC_IP'], zen, env['PUBLIC_IP']))

def run_domain_checks(rounded_time, env, output, pool, domains_to_check=None):
	# Get the list of domains we handle mail for.
	mail_domains = get_mail_domains(env)

	# Get the list of domains we serve DNS zones for (i.e. does not include subdomains).
	dns_zonefiles = dict(get_dns_zones(env))
	dns_domains = set(dns_zonefiles)

	# Get the list of domains we serve HTTPS for.
	web_domains = set(get_web_domains(env))

	if domains_to_check is None:
		domains_to_check = mail_domains | dns_domains | web_domains

	# Remove "www", "autoconfig", "autodiscover", and "mta-sts" subdomains, which we group with their parent,
	# if their parent is in the domains to check list.
	domains_to_check = [
		d for d in domains_to_check
		if not (
		   d.split(".", 1)[0] in {"www", "autoconfig", "autodiscover", "mta-sts"}
		   and len(d.split(".", 1)) == 2
		   and d.split(".", 1)[1] in domains_to_check
		)
	]

	# Get the list of domains that we don't serve web for because of a custom CNAME/A record.
	domains_with_a_records = get_domains_with_a_records(env)

	# Serial version:
	#for domain in sort_domains(domains_to_check, env):
	#	run_domain_checks_on_domain(domain, rounded_time, env, dns_domains, dns_zonefiles, mail_domains, web_domains)

	# Parallelize the checks across a worker pool.
	args = ((domain, rounded_time, env, dns_domains, dns_zonefiles, mail_domains, web_domains, domains_with_a_records)
		for domain in domains_to_check)
	ret = pool.starmap(run_domain_checks_on_domain, args, chunksize=1)
	ret = dict(ret) # (domain, output) => { domain: output }
	for domain in sort_domains(ret, env):
		ret[domain].playback(output)

def run_domain_checks_on_domain(domain, rounded_time, env, dns_domains, dns_zonefiles, mail_domains, web_domains, domains_with_a_records):
	output = BufferedOutput()

	# When running inside Flask, the worker threads don't get a thread pool automatically.
	# Also this method is called in a forked worker pool, so creating a new loop is probably
	# a good idea.
	asyncio.set_event_loop(asyncio.new_event_loop())

	# we'd move this up, but this returns non-pickleable values
	ssl_certificates = get_ssl_certificates(env)

	# The domain is IDNA-encoded in the database, but for display use Unicode.
	try:
		domain_display = idna.decode(domain.encode('ascii'))
		output.add_heading(domain_display)
	except (ValueError, UnicodeError, idna.IDNAError) as e:
		# Looks like we have some invalid data in our database.
		output.add_heading(domain)
		output.print_error("Domain name is invalid: " + str(e))

	if domain == env["PRIMARY_HOSTNAME"]:
		check_primary_hostname_dns(domain, env, output, dns_domains, dns_zonefiles)

	if domain in dns_domains:
		check_dns_zone(domain, env, output, dns_zonefiles)

	if domain in mail_domains:
		check_mail_domain(domain, env, output)

	if domain in web_domains:
		check_web_domain(domain, rounded_time, ssl_certificates, env, output)

	if domain in dns_domains:
		check_dns_zone_suggestions(domain, env, output, dns_zonefiles, domains_with_a_records)

	# Check auto-configured subdomains. See run_domain_checks.
	# Skip mta-sts because we check the policy directly.
	for label in ("www", "autoconfig", "autodiscover"):
		subdomain = label + "." + domain
		if subdomain in web_domains or subdomain in mail_domains:
			# Run checks.
			subdomain_output = run_domain_checks_on_domain(subdomain, rounded_time, env, dns_domains, dns_zonefiles, mail_domains, web_domains, domains_with_a_records)

			# Prepend the domain name to the start of each check line, and then add to the
			# checks for this domain.
			for attr, args, kwargs in subdomain_output[1].buf:
				if attr == "add_heading":
					# Drop the heading, but use its text as the subdomain name in
					# each line since it is in Unicode form.
					subdomain = args[0]
					continue
				if len(args) == 1 and isinstance(args[0], str):
					args = [ subdomain + ": " + args[0] ]
				getattr(output, attr)(*args, **kwargs)

	return (domain, output)

def check_primary_hostname_dns(domain, env, output, dns_domains, dns_zonefiles):
	# If a DS record is set on the zone containing this domain, check DNSSEC now.
	has_dnssec = False
	for zone in dns_domains:
		if (zone == domain or domain.endswith("." + zone)) and query_dns(zone, "DS", nxdomain=None) is not None:
			has_dnssec = True
			check_dnssec(zone, env, output, dns_zonefiles, is_checking_primary=True)

	ip = query_dns(domain, "A")
	ns_ips = query_dns("ns1." + domain, "A") + '/' + query_dns("ns2." + domain, "A")
	my_ips = env['PUBLIC_IP'] + ((" / "+env['PUBLIC_IPV6']) if env.get("PUBLIC_IPV6") else "")

	# Check that the ns1/ns2 hostnames resolve to A records. This information probably
	# comes from the TLD since the information is set at the registrar as glue records.
	# We're probably not actually checking that here but instead checking that we, as
	# the nameserver, are reporting the right info --- but if the glue is incorrect this
	# will probably fail.
	if ns_ips == env['PUBLIC_IP'] + '/' + env['PUBLIC_IP']:
		output.print_ok("Nameserver glue records are correct at registrar. [ns1/ns2.{} ↦ {}]".format(env['PRIMARY_HOSTNAME'], env['PUBLIC_IP']))

	elif ip == env['PUBLIC_IP']:
		# The NS records are not what we expect, but the domain resolves correctly, so
		# the user may have set up external DNS. List this discrepancy as a warning.
		output.print_warning("""Nameserver glue records (ns1.{} and ns2.{}) should be configured at your domain name
			registrar as having the IP address of this box ({}). They currently report addresses of {}. If you have set up External DNS, this may be OK.""".format(env['PRIMARY_HOSTNAME'], env['PRIMARY_HOSTNAME'], env['PUBLIC_IP'], ns_ips))

	else:
		output.print_error("""Nameserver glue records are incorrect. The ns1.{} and ns2.{} nameservers must be configured at your domain name
			registrar as having the IP address {}. They currently report addresses of {}. It may take several hours for
			public DNS to update after a change.""".format(env['PRIMARY_HOSTNAME'], env['PRIMARY_HOSTNAME'], env['PUBLIC_IP'], ns_ips))

	# Check that PRIMARY_HOSTNAME resolves to PUBLIC_IP[V6] in public DNS.
	ipv6 = query_dns(domain, "AAAA") if env.get("PUBLIC_IPV6") else None
	if ip == env['PUBLIC_IP'] and not (ipv6 and env['PUBLIC_IPV6'] and ipv6 != normalize_ip(env['PUBLIC_IPV6'])):
		output.print_ok("Domain resolves to box's IP address. [{} ↦ {}]".format(env['PRIMARY_HOSTNAME'], my_ips))
	else:
		output.print_error("""This domain must resolve to this box's IP address ({}) in public DNS but it currently resolves
			to {}. It may take several hours for public DNS to update after a change. This problem may result from other
			issues listed above.""".format(my_ips, ip + ((" / " + ipv6) if ipv6 is not None else "")))


	# Check reverse DNS matches the PRIMARY_HOSTNAME. Note that it might not be
	# a DNS zone if it is a subdomain of another domain we have a zone for.
	existing_rdns_v4 = query_dns(dns.reversename.from_address(env['PUBLIC_IP']), "PTR")
	existing_rdns_v6 = query_dns(dns.reversename.from_address(env['PUBLIC_IPV6']), "PTR") if env.get("PUBLIC_IPV6") else None
	if existing_rdns_v4 == domain and existing_rdns_v6 in {None, domain}:
		output.print_ok("Reverse DNS is set correctly at ISP. [{} ↦ {}]".format(my_ips, env['PRIMARY_HOSTNAME']))
	elif existing_rdns_v4 == existing_rdns_v6 or existing_rdns_v6 is None:
		output.print_error(f"""This box's reverse DNS is currently {existing_rdns_v4}, but it should be {domain}. Your ISP or cloud provider will have instructions
			on setting up reverse DNS for this box.""" )
	else:
		output.print_error(f"""This box's reverse DNS is currently {existing_rdns_v4} (IPv4) and {existing_rdns_v6} (IPv6), but it should be {domain}. Your ISP or cloud provider will have instructions
			on setting up reverse DNS for this box.""" )

	# Check the TLSA record.
	tlsa_qname = "_25._tcp." + domain
	tlsa25 = query_dns(tlsa_qname, "TLSA", nxdomain=None)
	tlsa25_expected = build_tlsa_record(env)
	if tlsa25 == tlsa25_expected:
		output.print_ok("""The DANE TLSA record for incoming mail is correct (%s).""" % tlsa_qname,)
	elif tlsa25 is None:
		if has_dnssec:
			# Omit a warning about it not being set if DNSSEC isn't enabled,
			# since TLSA shouldn't be used without DNSSEC.
			output.print_warning("""The DANE TLSA record for incoming mail is not set. This is optional.""")
	else:
		output.print_error(f"""The DANE TLSA record for incoming mail ({tlsa_qname}) is not correct. It is '{tlsa25}' but it should be '{tlsa25_expected}'.
			It may take several hours for public DNS to update after a change.""")

	# Check that the hostmaster@ email address exists.
	check_alias_exists("Hostmaster contact address", "hostmaster@" + domain, env, output)

def check_alias_exists(alias_name, alias, env, output):
	mail_aliases = {address: receivers for address, receivers, *_ in get_mail_aliases(env)}
	if alias in mail_aliases:
		if mail_aliases[alias]:
			output.print_ok(f"{alias_name} exists as a mail alias. [{alias} ↦ {mail_aliases[alias]}]")
		else:
			output.print_error("""You must set the destination of the mail alias for %s to direct email to you or another administrator.""" % alias)
	else:
		output.print_error("""You must add a mail alias for %s which directs email to you or another administrator.""" % alias)

def check_dns_zone(domain, env, output, dns_zonefiles):
	# If a DS record is set at the registrar, check DNSSEC first because it will affect the NS query.
	# If it is not set, we suggest it last.
	if query_dns(domain, "DS", nxdomain=None) is not None:
		check_dnssec(domain, env, output, dns_zonefiles)

	# We provide a DNS zone for the domain. It should have NS records set up
	# at the domain name's registrar pointing to this box. The secondary DNS
	# server may be customized.
	# (I'm not sure whether this necessarily tests the TLD's configuration,
	# as it should, or if one successful NS line at the TLD will result in
	# this query being answered by the box, which would mean the test is only
	# half working.)

	custom_dns_records = list(get_custom_dns_config(env)) # generator => list so we can reuse it
	correct_ip = "; ".join(sorted(get_custom_dns_records(custom_dns_records, domain, "A"))) or env['PUBLIC_IP']
	custom_secondary_ns = get_secondary_dns(custom_dns_records, mode="NS")
	secondary_ns = custom_secondary_ns or ["ns2." + env['PRIMARY_HOSTNAME']]

	existing_ns = query_dns(domain, "NS")
	correct_ns = "; ".join(sorted(["ns1." + env["PRIMARY_HOSTNAME"], *secondary_ns]))
	ip = query_dns(domain, "A")

	probably_external_dns = False

	if existing_ns.lower() == correct_ns.lower():
		output.print_ok("Nameservers are set correctly at registrar. [%s]" % correct_ns)
	elif ip == correct_ip:
		# The domain resolves correctly, so maybe the user is using External DNS.
		output.print_warning(f"""The nameservers set on this domain at your domain name registrar should be {correct_ns}. They are currently {existing_ns}.
			If you are using External DNS, this may be OK.""" )
		probably_external_dns = True
	else:
		output.print_error(f"""The nameservers set on this domain are incorrect. They are currently {existing_ns}. Use your domain name registrar's
			control panel to set the nameservers to {correct_ns}.""" )

	# Check that each custom secondary nameserver resolves the IP address.

	if custom_secondary_ns and not probably_external_dns:
		for ns in custom_secondary_ns:
			# We must first resolve the nameserver to an IP address so we can query it.
			ns_ips = query_dns(ns, "A")
			if not ns_ips or ns_ips in {'[Not Set]', '[timeout]'}:
				output.print_error("Secondary nameserver %s is not valid (it doesn't resolve to an IP address)." % ns)
				continue
			# Choose the first IP if nameserver returns multiple
			ns_ip = ns_ips.split('; ')[0]

			# Now query it to see what it says about this domain.
			ip = query_dns(domain, "A", at=ns_ip, nxdomain=None)
			if ip == correct_ip:
				output.print_ok("Secondary nameserver %s resolved the domain correctly." % ns)
			elif ip is None:
				output.print_error("Secondary nameserver %s is not configured to resolve this domain." % ns)
			else:
				output.print_error(f"Secondary nameserver {ns} is not configured correctly. (It resolved this domain as {ip}. It should be {correct_ip}.)")

def check_dns_zone_suggestions(domain, env, output, dns_zonefiles, domains_with_a_records):
	# Warn if a custom DNS record is preventing this or the automatic www redirect from
	# being served.
	if domain in domains_with_a_records:
		output.print_warning("""Web has been disabled for this domain because you have set a custom DNS record.""")
	if "www." + domain in domains_with_a_records:
		output.print_warning("""A redirect from 'www.%s' has been disabled for this domain because you have set a custom DNS record on the www subdomain.""" % domain)

	# Since DNSSEC is optional, if a DS record is NOT set at the registrar suggest it.
	# (If it was set, we did the check earlier.)
	if query_dns(domain, "DS", nxdomain=None) is None:
		check_dnssec(domain, env, output, dns_zonefiles)


def check_dnssec(domain, env, output, dns_zonefiles, is_checking_primary=False):
	# See if the domain has a DS record set at the registrar. The DS record must
	# match one of the keys that we've used to sign the zone. It may use one of
	# several hashing algorithms. We've pre-generated all possible valid DS
	# records, although some will be preferred.

	alg_name_map = { '7': 'RSASHA1-NSEC3-SHA1', '8': 'RSASHA256', '13': 'ECDSAP256SHA256' }
	digalg_name_map = { '1': 'SHA-1', '2': 'SHA-256', '4': 'SHA-384' }

	# Read in the pre-generated DS records
	expected_ds_records = { }
	ds_file = '/etc/nsd/zones/' + dns_zonefiles[domain] + '.ds'
	if not os.path.exists(ds_file): return # Domain is in our database but DNS has not yet been updated.
	with open(ds_file, encoding="utf-8") as f:
		for rr_ds in f:
			rr_ds = rr_ds.rstrip()
			ds_keytag, ds_alg, ds_digalg, ds_digest = rr_ds.split("\t")[4].split(" ")

			# Some registrars may want the public key so they can compute the digest. The DS
			# record that we suggest using is for the KSK (and that's how the DS records were generated).
			# We'll also give the nice name for the key algorithm.
			dnssec_keys = load_env_vars_from_file(os.path.join(env['STORAGE_ROOT'], 'dns/dnssec/%s.conf' % alg_name_map[ds_alg]))
			with open(os.path.join(env['STORAGE_ROOT'], 'dns/dnssec/' + dnssec_keys['KSK'] + '.key'), encoding="utf-8") as f:
				dnsssec_pubkey = f.read().split("\t")[3].split(" ")[3]

			expected_ds_records[ (ds_keytag, ds_alg, ds_digalg, ds_digest) ] = {
				"record": rr_ds,
				"keytag": ds_keytag,
				"alg": ds_alg,
				"alg_name": alg_name_map[ds_alg],
				"digalg": ds_digalg,
				"digalg_name": digalg_name_map[ds_digalg],
				"digest": ds_digest,
				"pubkey": dnsssec_pubkey,
			}

	# Query public DNS for the DS record at the registrar.
	ds = query_dns(domain, "DS", nxdomain=None, as_list=True)
	if ds is None or isinstance(ds, str): ds = []

	# There may be more that one record, so we get the result as a list.
	# Filter out records that don't look valid, just in case, and split
	# each record on spaces.
	ds = [tuple(str(rr).split(" ")) for rr in ds if len(str(rr).split(" ")) == 4]

	if len(ds) == 0:
		output.print_warning("""This domain's DNSSEC DS record is not set. The DS record is optional. The DS record activates DNSSEC. See below for instructions.""")
	else:
		matched_ds = set(ds) & set(expected_ds_records)
		if matched_ds:
			# At least one DS record matches one that corresponds with one of the ways we signed
			# the zone, so it is valid.
			#
			# But it may not be preferred. Only algorithm 13 is preferred. Warn if any of the
			# matched zones uses a different algorithm.
			if {r[1] for r in matched_ds} == { '13' } and {r[2] for r in matched_ds} <= { '2', '4' }: # all are alg 13 and digest type 2 or 4
				output.print_ok("DNSSEC 'DS' record is set correctly at registrar.")
				return
			elif len([r for r in matched_ds if r[1] == '13' and r[2] in { '2', '4' }]) > 0: # some but not all are alg 13
				output.print_ok("DNSSEC 'DS' record is set correctly at registrar. (Records using algorithm other than ECDSAP256SHA256 and digest types other than SHA-256/384 should be removed.)")
				return
			else: # no record uses alg 13
				output.print_warning("""DNSSEC 'DS' record set at registrar is valid but should be updated to ECDSAP256SHA256 and SHA-256 (see below).
				IMPORTANT: Do not delete existing DNSSEC 'DS' records for this domain until confirmation that the new DNSSEC 'DS' record
				for this domain is valid.""")
		else:
			if is_checking_primary:
				output.print_error("""The DNSSEC 'DS' record for %s is incorrect. See further details below.""" % domain)
				return
			output.print_error("""This domain's DNSSEC DS record is incorrect. The chain of trust is broken between the public DNS system
				and this machine's DNS server. It may take several hours for public DNS to update after a change. If you did not recently
				make a change, you must resolve this immediately (see below).""")

	output.print_line("""Follow the instructions provided by your domain name registrar to set a DS record.
		Registrars support different sorts of DS records. Use the first option that works:""")
	preferred_ds_order = [(7, 2), (8, 4), (13, 4), (8, 2), (13, 2)] # low to high, see https://github.com/mail-in-a-box/mailinabox/issues/1998

	def preferred_ds_order_func(ds_suggestion):
		k = (int(ds_suggestion['alg']), int(ds_suggestion['digalg']))
		if k in preferred_ds_order:
			return preferred_ds_order.index(k)
		return -1 # index before first item
	output.print_line("")
	for i, ds_suggestion in enumerate(sorted(expected_ds_records.values(), key=preferred_ds_order_func, reverse=True)):
		if preferred_ds_order_func(ds_suggestion) == -1: continue # don't offer record types that the RFC says we must not offer
		output.print_line("")
		output.print_line("Option " + str(i+1) + ":")
		output.print_line("----------")
		output.print_line("Key Tag: " + ds_suggestion['keytag'])
		output.print_line("Key Flags: KSK / 257")
		output.print_line("Algorithm: {} / {}".format(ds_suggestion['alg'], ds_suggestion['alg_name']))
		output.print_line("Digest Type: {} / {}".format(ds_suggestion['digalg'], ds_suggestion['digalg_name']))
		output.print_line("Digest: " + ds_suggestion['digest'])
		output.print_line("Public Key: ")
		output.print_line(ds_suggestion['pubkey'], monospace=True)
		output.print_line("")
		output.print_line("Bulk/Record Format:")
		output.print_line(ds_suggestion['record'], monospace=True)
	if len(ds) > 0:
		output.print_line("")
		output.print_line("The DS record is currently set to:")
		for rr in sorted(ds):
			output.print_line("Key Tag: {}, Algorithm: {}, Digest Type: {}, Digest: {}".format(*rr))

def check_mail_domain(domain, env, output):
	# Check the MX record.

	recommended_mx = "10 " + env['PRIMARY_HOSTNAME']
	mx = query_dns(domain, "MX", nxdomain=None)

	if mx is None or mx == "[timeout]":
		mxhost = None
	else:
		# query_dns returns a semicolon-delimited list
		# of priority-host pairs.
		mxhost = mx.split('; ')[0].split(' ')[1]

	if mxhost is None:
		# A missing MX record is okay on the primary hostname because
		# the primary hostname's A record (the MX fallback) is... itself,
		# which is what we want the MX to be.
		if domain == env['PRIMARY_HOSTNAME']:
			output.print_ok(f"Domain's email is directed to this domain. [{domain} has no MX record, which is ok]")

		# And a missing MX record is okay on other domains if the A record
		# matches the A record of the PRIMARY_HOSTNAME. Actually this will
		# probably confuse DANE TLSA, but we'll let that slide for now.
		else:
			domain_a = query_dns(domain, "A", nxdomain=None)
			primary_a = query_dns(env['PRIMARY_HOSTNAME'], "A", nxdomain=None)
			if domain_a is not None and domain_a == primary_a:
				output.print_ok(f"Domain's email is directed to this domain. [{domain} has no MX record but its A record is OK]")
			else:
				output.print_error(f"""This domain's DNS MX record is not set. It should be '{recommended_mx}'. Mail will not
					be delivered to this box. It may take several hours for public DNS to update after a
					change. This problem may result from other issues listed here.""")

	elif mxhost == env['PRIMARY_HOSTNAME']:
		good_news = f"Domain's email is directed to this domain. [{domain} ↦ {mx}]"
		if mx != recommended_mx:
			good_news += f"  This configuration is non-standard.  The recommended configuration is '{recommended_mx}'."
		output.print_ok(good_news)

		# Check MTA-STS policy.
		loop = asyncio.new_event_loop()
		sts_resolver = postfix_mta_sts_resolver.resolver.STSResolver(loop=loop)
		valid, policy = loop.run_until_complete(sts_resolver.resolve(domain))
		if valid == postfix_mta_sts_resolver.resolver.STSFetchResult.VALID:
			if policy[1].get("mx") == [env['PRIMARY_HOSTNAME']] and policy[1].get("mode") == "enforce": # policy[0] is the policyid
				output.print_ok("MTA-STS policy is present.")
			else:
				output.print_error(f"MTA-STS policy is present but has unexpected settings. [{policy[1]}]")
		else:
			output.print_error(f"MTA-STS policy is missing: {valid}")

	else:
		output.print_error(f"""This domain's DNS MX record is incorrect. It is currently set to '{mx}' but should be '{recommended_mx}'. Mail will not
			be delivered to this box. It may take several hours for public DNS to update after a change. This problem may result from
			other issues listed here.""")

	# Check that the postmaster@ email address exists. Not required if the domain has a
	# catch-all address or domain alias.
	if "@" + domain not in [address for address, *_ in get_mail_aliases(env)]:
		check_alias_exists("Postmaster contact address", "postmaster@" + domain, env, output)

	# Stop if the domain is listed in the Spamhaus Domain Block List.
	# The user might have chosen a domain that was previously in use by a spammer
	# and will not be able to reliably send mail.
	
	# See https://www.spamhaus.org/news/article/807/using-our-public-mirrors-check-your-return-codes-now. for
	# information on spamhaus return codes
	dbl = query_dns(domain+'.dbl.spamhaus.org', "A", nxdomain=None)
	if dbl is None:
		output.print_ok("Domain is not blacklisted by dbl.spamhaus.org.")
	elif dbl == "[timeout]":
		output.print_warning(f"Connection to dbl.spamhaus.org timed out. Could not determine whether the domain {domain} is blacklisted. Please try again later.")
	elif dbl == "[Not Set]":
		output.print_warning(f"Could not connect to dbl.spamhaus.org. Could not determine whether the domain {domain} is blacklisted. Please try again later.")
	elif dbl == "127.255.255.252":
		output.print_warning("Incorrect spamhaus query: %s. Could not determine whether the domain %s is blacklisted." % (domain+'.dbl.spamhaus.org', domain))
	elif dbl == "127.255.255.254":
		output.print_warning("Mail-in-a-Box is configured to use a public DNS server. This is not supported by spamhaus. Could not determine whether the domain {} is blacklisted.".format(domain))
	elif dbl == "127.255.255.255":
		output.print_warning("Too many queries have been performed on the spamhaus server. Could not determine whether the domain {} is blacklisted.".format(domain))
	else:
		output.print_error(f"""This domain is listed in the Spamhaus Domain Block List (code {dbl}),
			which may prevent recipients from receiving your mail.
			See http://www.spamhaus.org/dbl/ and http://www.spamhaus.org/query/domain/{domain}.""")

def check_web_domain(domain, rounded_time, ssl_certificates, env, output):
	# See if the domain's A record resolves to our PUBLIC_IP. This is already checked
	# for PRIMARY_HOSTNAME, for which it is required for mail specifically. For it and
	# other domains, it is required to access its website.
	if domain != env['PRIMARY_HOSTNAME']:
		ok_values = []
		for (rtype, expected) in (("A", env['PUBLIC_IP']), ("AAAA", env.get('PUBLIC_IPV6'))):
			if not expected: continue # IPv6 is not configured
			value = query_dns(domain, rtype)
			if value == normalize_ip(expected):
				ok_values.append(value)
			else:
				output.print_error(f"""This domain should resolve to this box's IP address ({rtype} {expected}) if you would like the box to serve
					webmail or a website on this domain. The domain currently resolves to {value} in public DNS. It may take several hours for
					public DNS to update after a change. This problem may result from other issues listed here.""")
				return

		# If both A and AAAA are correct...
		output.print_ok("Domain resolves to this box's IP address. [{} ↦ {}]".format(domain, '; '.join(ok_values)))


	# We need a TLS certificate for PRIMARY_HOSTNAME because that's where the
	# user will log in with IMAP or webmail. Any other domain we serve a
	# website for also needs a signed certificate.
	check_ssl_cert(domain, rounded_time, ssl_certificates, env, output)

def query_dns(qname, rtype, nxdomain='[Not Set]', at=None, as_list=False):
	# Make the qname absolute by appending a period. Without this, dns.resolver.query
	# will fall back a failed lookup to a second query with this machine's hostname
	# appended. This has been causing some false-positive Spamhaus reports. The
	# reverse DNS lookup will pass a dns.name.Name instance which is already
	# absolute so we should not modify that.
	if isinstance(qname, str):
		qname += "."

	# Use the default nameservers (as defined by the system, which is our locally
	# running bind server), or if the 'at' argument is specified, use that host
	# as the nameserver.
	resolver = dns.resolver.get_default_resolver()

	# Make sure at is not a string that cannot be used as a nameserver
	if at and at not in {'[Not set]', '[timeout]'}:
		resolver = dns.resolver.Resolver()
		resolver.nameservers = [at]

	# Set a timeout so that a non-responsive server doesn't hold us back.
	resolver.timeout = 5
	# The number of seconds to spend trying to get an answer to the question. If the
	# lifetime expires a dns.exception.Timeout exception will be raised.
	resolver.lifetime = 5

	# Do the query.
	try:
		response = resolver.resolve(qname, rtype)
	except (dns.resolver.NoNameservers, dns.resolver.NXDOMAIN, dns.resolver.NoAnswer):
		# Host did not have an answer for this query; not sure what the
		# difference is between the two exceptions.
		return nxdomain
	except dns.exception.Timeout:
		return "[timeout]"

	# Normalize IP addresses. IP address --- especially IPv6 addresses --- can
	# be expressed in equivalent string forms. Canonicalize the form before
	# returning them. The caller should normalize any IP addresses the result
	# of this method is compared with.
	if rtype in {"A", "AAAA"}:
		response = [normalize_ip(str(r)) for r in response]

	if as_list:
		return response

	# There may be multiple answers; concatenate the response. Remove trailing
	# periods from responses since that's how qnames are encoded in DNS but is
	# confusing for us. The order of the answers doesn't matter, so sort so we
	# can compare to a well known order.
	return "; ".join(sorted(str(r).rstrip('.') for r in response))

def check_ssl_cert(domain, rounded_time, ssl_certificates, env, output):
	# Check that TLS certificate is signed.

	# Skip the check if the A record is not pointed here.
	if query_dns(domain, "A", None) not in {env['PUBLIC_IP'], None}: return

	# Where is the certificate file stored?
	tls_cert = get_domain_ssl_files(domain, ssl_certificates, env, allow_missing_cert=True)
	if tls_cert is None:
		output.print_warning("""No TLS (SSL) certificate is installed for this domain. Visitors to a website on
			this domain will get a security warning. If you are not serving a website on this domain, you do
			not need to take any action. Use the TLS Certificates page in the control panel to install a
			TLS certificate.""")
		return

	# Check that the certificate is good.

	cert_status, cert_status_details = check_certificate(domain, tls_cert["certificate"], tls_cert["private-key"], rounded_time=rounded_time)

	if cert_status == "OK":
		# The certificate is ok. The details has expiry info.
		output.print_ok("TLS (SSL) certificate is signed & valid. " + cert_status_details)

	elif cert_status == "SELF-SIGNED":
		# Offer instructions for purchasing a signed certificate.
		if domain == env['PRIMARY_HOSTNAME']:
			output.print_error("""The TLS (SSL) certificate for this domain is currently self-signed. You will get a security
			warning when you check or send email and when visiting this domain in a web browser (for webmail or
			static site hosting).""")
		else:
			output.print_error("""The TLS (SSL) certificate for this domain is self-signed.""")

	else:
		output.print_error("The TLS (SSL) certificate has a problem: " + cert_status)
		if cert_status_details:
			output.print_line("")
			output.print_line(cert_status_details)
			output.print_line("")

_apt_updates = None
def list_apt_updates(apt_update=True):
	# See if we have this information cached recently.
	# Keep the information for 8 hours.
	global _apt_updates
	if _apt_updates is not None and _apt_updates[0] > datetime.datetime.now() - datetime.timedelta(hours=8):
		return _apt_updates[1]

	# Run apt-get update to refresh package list. This should be running daily
	# anyway, so on the status checks page don't do this because it is slow.
	if apt_update:
		shell("check_call", ["/usr/bin/apt-get", "-qq", "update"])

	# Run apt-get upgrade in simulate mode to get a list of what
	# it would do.
	simulated_install = shell("check_output", ["/usr/bin/apt-get", "-qq", "-s", "upgrade"])
	pkgs = []
	for line in simulated_install.split('\n'):
		if line.strip() == "":
			continue
		if re.match(r'^Conf .*', line):
			 # remove these lines, not informative
			continue
		m = re.match(r'^Inst (.*) \[(.*)\] \((\S*)', line)
		if m:
			pkgs.append({ "package": m.group(1), "version": m.group(3), "current_version": m.group(2) })
		else:
			pkgs.append({ "package": "[" + line + "]", "version": "", "current_version": "" })

	# Cache for future requests.
	_apt_updates = (datetime.datetime.now(), pkgs)

	return pkgs

def what_version_is_this(env):
	# This function runs `git describe --always --abbrev=0` on the Mail-in-a-Box installation directory.
	# Git may not be installed and Mail-in-a-Box may not have been cloned from github,
	# so this function may raise all sorts of exceptions.
	miab_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
	return shell("check_output", ["/usr/bin/git", "describe", "--always", "--abbrev=0"], env={"GIT_DIR": os.path.join(miab_dir, '.git')}).strip()

def get_latest_miab_version():
	# This pings https://mailinabox.email/setup.sh and extracts the tag named in
	# the script to determine the current product version.
    from urllib.request import urlopen, HTTPError, URLError

    try:
        return re.search(b'TAG=(.*)', urlopen("https://mailinabox.email/setup.sh?ping=1", timeout=5).read()).group(1).decode("utf8")
    except (TimeoutError, HTTPError, URLError):
        return None

def check_miab_version(env, output):
	config = load_settings(env)

	try:
		this_ver = what_version_is_this(env)
	except:
		this_ver = "Unknown"

	if config.get("privacy", True):
		output.print_warning("You are running version Mail-in-a-Box %s. Mail-in-a-Box version check disabled by privacy setting." % this_ver)
	else:
		latest_ver = get_latest_miab_version()

		if this_ver == latest_ver:
			output.print_ok("Mail-in-a-Box is up to date. You are running version %s." % this_ver)
		elif latest_ver is None:
			output.print_error("Latest Mail-in-a-Box version could not be determined. You are running version %s." % this_ver)
		else:
			output.print_error(f"A new version of Mail-in-a-Box is available. You are running version {this_ver}. The latest version is {latest_ver}. For upgrade instructions, see https://mailinabox.email. ")

def run_and_output_changes(env, pool):
	import json
	from difflib import SequenceMatcher

	out = ConsoleOutput()

	# Run status checks.
	cur = BufferedOutput()
	run_checks(True, env, cur, pool)

	# Load previously saved status checks.
	cache_fn = "/var/cache/mailinabox/status_checks.json"
	if os.path.exists(cache_fn):
		with open(cache_fn, encoding="utf-8") as f:
			try:
				prev = json.load(f)
			except json.JSONDecodeError:
				prev = []

		# Group the serial output into categories by the headings.
		def group_by_heading(lines):
			from collections import OrderedDict
			ret = OrderedDict()
			k = []
			ret["No Category"] = k
			for line_type, line_args, line_kwargs in lines:
				if line_type == "add_heading":
					k = []
					ret[line_args[0]] = k
				else:
					k.append((line_type, line_args, line_kwargs))
			return ret
		prev_status = group_by_heading(prev)
		cur_status = group_by_heading(cur.buf)

		# Compare the previous to the current status checks
		# category by category.
		for category, cur_lines in cur_status.items():
			if category not in prev_status:
				out.add_heading(category + " -- Added")
				BufferedOutput(with_lines=cur_lines).playback(out)
			else:
				# Actual comparison starts here...
				prev_lines = prev_status[category]
				def stringify(lines):
					return [json.dumps(line) for line in lines]
				diff = SequenceMatcher(None, stringify(prev_lines), stringify(cur_lines)).get_opcodes()
				for op, i1, i2, j1, j2 in diff:
					if op == "replace":
						out.add_heading(category + " -- Previously:")
					elif op == "delete":
						out.add_heading(category + " -- Removed")
					if op in {"replace", "delete"}:
						BufferedOutput(with_lines=prev_lines[i1:i2]).playback(out)

					if op == "replace":
						out.add_heading(category + " -- Currently:")
					elif op == "insert":
						out.add_heading(category + " -- Added")
					if op in {"replace", "insert"}:
						BufferedOutput(with_lines=cur_lines[j1:j2]).playback(out)

		for category, prev_lines in prev_status.items():
			if category not in cur_status:
				out.add_heading(category)
				out.print_warning("This section was removed.")

	# Store the current status checks output for next time.
	os.makedirs(os.path.dirname(cache_fn), exist_ok=True)
	with open(cache_fn, "w", encoding="utf-8") as f:
		json.dump(cur.buf, f, indent=True)

def normalize_ip(ip):
	# Use ipaddress module to normalize the IPv6 notation and
	# ensure we are matching IPv6 addresses written in different
	# representations according to rfc5952.
	import ipaddress
	try:
		return str(ipaddress.ip_address(ip))
	except:
		return ip

class FileOutput:
	def __init__(self, buf, width):
		self.buf = buf
		self.width = width

	def add_heading(self, heading):
		print(file=self.buf)
		print(heading, file=self.buf)
		print("=" * len(heading), file=self.buf)

	def print_ok(self, message):
		self.print_block(message, first_line="✓  ")

	def print_error(self, message):
		self.print_block(message, first_line="✖  ")

	def print_warning(self, message):
		self.print_block(message, first_line="?  ")

	def print_block(self, message, first_line="   "):
		print(first_line, end='', file=self.buf)
		message = re.sub("\n\\s*", " ", message)
		words = re.split(r"(\s+)", message)
		linelen = 0
		for w in words:
			if self.width and (linelen + len(w) > self.width-1-len(first_line)):
				print(file=self.buf)
				print("   ", end="", file=self.buf)
				linelen = 0
			if linelen == 0 and w.strip() == "": continue
			print(w, end="", file=self.buf)
			linelen += len(w)
		print(file=self.buf)

	def print_line(self, message, monospace=False):
		for line in message.split("\n"):
			self.print_block(line)

class ConsoleOutput(FileOutput):
	def __init__(self):
		self.buf = sys.stdout

		# Do nice line-wrapping according to the size of the terminal.
		# The 'stty' program queries standard input for terminal information.
		if sys.stdin.isatty():
			try:
				self.width = int(shell('check_output', ['stty', 'size']).split()[1])
			except:
				self.width = 76

		else:
			# However if standard input is not a terminal, we would get
			# "stty: standard input: Inappropriate ioctl for device". So
			# we test with sys.stdin.isatty first, and if it is not a
			# terminal don't do any line wrapping. When this script is
			# run from cron, or if stdin has been redirected, this happens.
			self.width = None

class BufferedOutput:
	# Record all of the instance method calls so we can play them back later.
	def __init__(self, with_lines=None):
		self.buf = with_lines if with_lines else []
	def __getattr__(self, attr):
		if attr not in {"add_heading", "print_ok", "print_error", "print_warning", "print_block", "print_line"}:
			raise AttributeError
		# Return a function that just records the call & arguments to our buffer.
		def w(*args, **kwargs):
			self.buf.append((attr, args, kwargs))
		return w
	def playback(self, output):
		for attr, args, kwargs in self.buf:
			getattr(output, attr)(*args, **kwargs)


if __name__ == "__main__":
	from utils import load_environment

	env = load_environment()

	if len(sys.argv) == 1:
		with multiprocessing.pool.Pool(processes=10) as pool:
			run_checks(False, env, ConsoleOutput(), pool)

	elif sys.argv[1] == "--show-changes":
		with multiprocessing.pool.Pool(processes=10) as pool:
			run_and_output_changes(env, pool)

	elif sys.argv[1] == "--check-primary-hostname":
		# See if the primary hostname appears resolvable and has a signed certificate.
		domain = env['PRIMARY_HOSTNAME']
		if query_dns(domain, "A") != env['PUBLIC_IP']:
			sys.exit(1)
		ssl_certificates = get_ssl_certificates(env)
		tls_cert = get_domain_ssl_files(domain, ssl_certificates, env)
		if not os.path.exists(tls_cert["certificate"]):
			sys.exit(1)
		cert_status, cert_status_details = check_certificate(domain, tls_cert["certificate"], tls_cert["private-key"], warn_if_expiring_soon=False)
		if cert_status != "OK":
			sys.exit(1)
		sys.exit(0)

	elif sys.argv[1] == "--version":
		print(what_version_is_this(env))

	elif sys.argv[1] == "--only":
		with multiprocessing.pool.Pool(processes=10) as pool:
			run_checks(False, env, ConsoleOutput(), pool, domains_to_check=sys.argv[2:])



File: management/dns_update.py

#!/usr/local/lib/mailinabox/env/bin/python

# Creates DNS zone files for all of the domains of all of the mail users
# and mail aliases and restarts nsd.
########################################################################

import sys, os, os.path, datetime, re, hashlib, base64
import ipaddress
import rtyaml
import dns.resolver

from utils import shell, load_env_vars_from_file, safe_domain_name, sort_domains, get_ssh_port
from ssl_certificates import get_ssl_certificates, check_certificate
import contextlib

# From https://stackoverflow.com/questions/3026957/how-to-validate-a-domain-name-using-regex-php/16491074#16491074
# This regular expression matches domain names according to RFCs, it also accepts fqdn with an leading dot,
# underscores, as well as asterisks which are allowed in domain names but not hostnames (i.e. allowed in
# DNS but not in URLs), which are common in certain record types like for DKIM.
DOMAIN_RE = r"^(?!\-)(?:[*][.])?(?:[a-zA-Z\d\-_]{0,62}[a-zA-Z\d_]\.){1,126}(?!\d+)[a-zA-Z\d_]{1,63}(\.?)$"

def get_dns_domains(env):
	# Add all domain names in use by email users and mail aliases, any
	# domains we serve web for (except www redirects because that would
	# lead to infinite recursion here) and ensure PRIMARY_HOSTNAME is in the list.
	from mailconfig import get_mail_domains
	from web_update import get_web_domains
	domains = set()
	domains |= set(get_mail_domains(env))
	domains |= set(get_web_domains(env, include_www_redirects=False))
	domains.add(env['PRIMARY_HOSTNAME'])
	return domains

def get_dns_zones(env):
	# What domains should we create DNS zones for? Never create a zone for
	# a domain & a subdomain of that domain.
	domains = get_dns_domains(env)

	# Exclude domains that are subdomains of other domains we know. Proceed
	# by looking at shorter domains first.
	zone_domains = set()
	for domain in sorted(domains, key=len):
		for d in zone_domains:
			if domain.endswith("." + d):
				# We found a parent domain already in the list.
				break
		else:
			# 'break' did not occur: there is no parent domain.
			zone_domains.add(domain)

	# Make a nice and safe filename for each domain.
	zonefiles = [[domain, safe_domain_name(domain) + ".txt"] for domain in zone_domains]

	# Sort the list so that the order is nice and so that nsd.conf has a
	# stable order so we don't rewrite the file & restart the service
	# meaninglessly.
	zone_order = sort_domains([ zone[0] for zone in zonefiles ], env)
	zonefiles.sort(key = lambda zone : zone_order.index(zone[0]) )

	return zonefiles

def do_dns_update(env, force=False):
	# Write zone files.
	os.makedirs('/etc/nsd/zones', exist_ok=True)
	zonefiles = []
	updated_domains = []
	for (domain, zonefile, records) in build_zones(env):
		# The final set of files will be signed.
		zonefiles.append((domain, zonefile + ".signed"))

		# See if the zone has changed, and if so update the serial number
		# and write the zone file.
		if not write_nsd_zone(domain, "/etc/nsd/zones/" + zonefile, records, env, force):
			# Zone was not updated. There were no changes.
			continue

		# Mark that we just updated this domain.
		updated_domains.append(domain)

		# Sign the zone.
		#
		# Every time we sign the zone we get a new result, which means
		# we can't sign a zone without bumping the zone's serial number.
		# Thus we only sign a zone if write_nsd_zone returned True
		# indicating the zone changed, and thus it got a new serial number.
		# write_nsd_zone is smart enough to check if a zone's signature
		# is nearing expiration and if so it'll bump the serial number
		# and return True so we get a chance to re-sign it.
		sign_zone(domain, zonefile, env)

	# Write the main nsd.conf file.
	if write_nsd_conf(zonefiles, list(get_custom_dns_config(env)), env):
		# Make sure updated_domains contains *something* if we wrote an updated
		# nsd.conf so that we know to restart nsd.
		if len(updated_domains) == 0:
			updated_domains.append("DNS configuration")

	# Tell nsd to reload changed zone files.
	if len(updated_domains) > 0:
		# 'reconfig' is needed if there are added or removed zones, but
		# it may not reload existing zones, so we call 'reload' too. If
		# nsd isn't running, nsd-control fails, so in that case revert
		# to restarting nsd to make sure it is running. Restarting nsd
		# should also refresh everything.
		try:
			shell('check_call', ["/usr/sbin/nsd-control", "reconfig"])
			shell('check_call', ["/usr/sbin/nsd-control", "reload"])
		except:
			shell('check_call', ["/usr/sbin/service", "nsd", "restart"])

	# Write the OpenDKIM configuration tables for all of the mail domains.
	from mailconfig import get_mail_domains
	if write_opendkim_tables(get_mail_domains(env), env):
		# Settings changed. Kick opendkim.
		shell('check_call', ["/usr/sbin/service", "opendkim", "restart"])
		if len(updated_domains) == 0:
			# If this is the only thing that changed?
			updated_domains.append("OpenDKIM configuration")

	# Clear bind9's DNS cache so our own DNS resolver is up to date.
	# (ignore errors with trap=True)
	shell('check_call', ["/usr/sbin/rndc", "flush"], trap=True)

	if len(updated_domains) == 0:
		# if nothing was updated (except maybe OpenDKIM's files), don't show any output
		return ""
	else:
		return "updated DNS: " + ",".join(updated_domains) + "\n"

########################################################################

def build_zones(env):
	# What domains (and their zone filenames) should we build?
	domains = get_dns_domains(env)
	zonefiles = get_dns_zones(env)

	# Create a dictionary of domains to a set of attributes for each
	# domain, such as whether there are mail users at the domain.
	from mailconfig import get_mail_domains
	from web_update import get_web_domains
	mail_domains = set(get_mail_domains(env))
	mail_user_domains = set(get_mail_domains(env, users_only=True)) # i.e. will log in for mail, Nextcloud
	web_domains = set(get_web_domains(env))
	auto_domains = web_domains - set(get_web_domains(env, include_auto=False))
	domains |= auto_domains # www redirects not included in the initial list, see above

	# Add ns1/ns2+PRIMARY_HOSTNAME which must also have A/AAAA records
	# when the box is acting as authoritative DNS server for its domains.
	for ns in ("ns1", "ns2"):
		d = ns + "." + env["PRIMARY_HOSTNAME"]
		domains.add(d)
		auto_domains.add(d)

	domains = {
		domain: {
			"user": domain in mail_user_domains,
			"mail": domain in mail_domains,
			"web": domain in web_domains,
			"auto": domain in auto_domains,
		}
		for domain in domains
	}

	# For MTA-STS, we'll need to check if the PRIMARY_HOSTNAME certificate is
	# singned and valid. Check that now rather than repeatedly for each domain.
	domains[env["PRIMARY_HOSTNAME"]]["certificate-is-valid"] = is_domain_cert_signed_and_valid(env["PRIMARY_HOSTNAME"], env)

	# Load custom records to add to zones.
	additional_records = list(get_custom_dns_config(env))

	# Build DNS records for each zone.
	for domain, zonefile in zonefiles:
		# Build the records to put in the zone.
		records = build_zone(domain, domains, additional_records, env)
		yield (domain, zonefile, records)

def build_zone(domain, domain_properties, additional_records, env, is_zone=True):
	records = []

	# For top-level zones, define the authoritative name servers.
	#
	# Normally we are our own nameservers. Some TLDs require two distinct IP addresses,
	# so we allow the user to override the second nameserver definition so that
	# secondary DNS can be set up elsewhere.
	#
	# 'False' in the tuple indicates these records would not be used if the zone
	# is managed outside of the box.
	if is_zone:
		# Obligatory NS record to ns1.PRIMARY_HOSTNAME.
		records.append((None,  "NS",  "ns1.%s." % env["PRIMARY_HOSTNAME"], False))

		# NS record to ns2.PRIMARY_HOSTNAME or whatever the user overrides.
		# User may provide one or more additional nameservers
		secondary_ns_list = get_secondary_dns(additional_records, mode="NS") \
			or ["ns2." + env["PRIMARY_HOSTNAME"]]
		records.extend((None,  "NS", secondary_ns+'.', False) for secondary_ns in secondary_ns_list)


	# In PRIMARY_HOSTNAME...
	if domain == env["PRIMARY_HOSTNAME"]:
		# Set the A/AAAA records. Do this early for the PRIMARY_HOSTNAME so that the user cannot override them
		# and we can provide different explanatory text.
		records.append((None, "A", env["PUBLIC_IP"], "Required. Sets the IP address of the box."))
		if env.get("PUBLIC_IPV6"): records.append((None, "AAAA", env["PUBLIC_IPV6"], "Required. Sets the IPv6 address of the box."))

		# Add a DANE TLSA record for SMTP.
		records.append(("_25._tcp", "TLSA", build_tlsa_record(env), "Recommended when DNSSEC is enabled. Advertises to mail servers connecting to the box that mandatory encryption should be used."))

		# Add a DANE TLSA record for HTTPS, which some browser extensions might make use of.
		records.append(("_443._tcp", "TLSA", build_tlsa_record(env), "Optional. When DNSSEC is enabled, provides out-of-band HTTPS certificate validation for a few web clients that support it."))

		# Add a SSHFP records to help SSH key validation. One per available SSH key on this system.
		records.extend((None, "SSHFP", value, "Optional. Provides an out-of-band method for verifying an SSH key before connecting. Use 'VerifyHostKeyDNS yes' (or 'VerifyHostKeyDNS ask') when connecting with ssh.") for value in build_sshfp_records())

	# Add DNS records for any subdomains of this domain. We should not have a zone for
	# both a domain and one of its subdomains.
	if is_zone: # don't recurse when we're just loading data for a subdomain
		subdomains = [d for d in domain_properties if d.endswith("." + domain)]
		for subdomain in subdomains:
			subdomain_qname = subdomain[0:-len("." + domain)]
			subzone = build_zone(subdomain, domain_properties, additional_records, env, is_zone=False)
			for child_qname, child_rtype, child_value, child_explanation in subzone:
				if child_qname is None:
					child_qname = subdomain_qname
				else:
					child_qname += "." + subdomain_qname
				records.append((child_qname, child_rtype, child_value, child_explanation))

	has_rec_base = list(records) # clone current state
	def has_rec(qname, rtype, prefix=None):
		return any(rec[0] == qname and rec[1] == rtype and (prefix is None or rec[2].startswith(prefix)) for rec in has_rec_base)

	# The user may set other records that don't conflict with our settings.
	# Don't put any TXT records above this line, or it'll prevent any custom TXT records.
	for qname, rtype, value in filter_custom_records(domain, additional_records):
		# Don't allow custom records for record types that override anything above.
		# But allow multiple custom records for the same rtype --- see how has_rec_base is used.
		if has_rec(qname, rtype): continue

		# The "local" keyword on A/AAAA records are short-hand for our own IP.
		# This also flags for web configuration that the user wants a website here.
		if rtype == "A" and value == "local":
			value = env["PUBLIC_IP"]
		if rtype == "AAAA" and value == "local":
			if "PUBLIC_IPV6" in env:
				value = env["PUBLIC_IPV6"]
			else:
				continue
		records.append((qname, rtype, value, "(Set by user.)"))

	# Add A/AAAA defaults if not overridden by the user's custom settings (and not otherwise configured).
	# Any CNAME or A record on the qname overrides A and AAAA. But when we set the default A record,
	# we should not cause the default AAAA record to be skipped because it thinks a custom A record
	# was set. So set has_rec_base to a clone of the current set of DNS settings, and don't update
	# during this process.
	has_rec_base = list(records)
	a_expl = "Required. May have a different value. Sets the IP address that %s resolves to for web hosting and other services besides mail. The A record must be present but its value does not affect mail delivery." % domain
	if domain_properties[domain]["auto"]:
		if domain.startswith(("ns1.", "ns2.")): a_expl = False # omit from 'External DNS' page since this only applies if box is its own DNS server
		if domain.startswith("www."): a_expl = "Optional. Sets the IP address that %s resolves to so that the box can provide a redirect to the parent domain." % domain
		if domain.startswith("mta-sts."): a_expl = "Optional. MTA-STS Policy Host serving /.well-known/mta-sts.txt."
		if domain.startswith("autoconfig."): a_expl = "Provides email configuration autodiscovery support for Thunderbird Autoconfig."
		if domain.startswith("autodiscover."): a_expl = "Provides email configuration autodiscovery support for Z-Push ActiveSync Autodiscover."
	defaults = [
		(None,  "A",    env["PUBLIC_IP"], a_expl),
		(None,  "AAAA", env.get('PUBLIC_IPV6'), "Optional. Sets the IPv6 address that %s resolves to, e.g. for web hosting. (It is not necessary for receiving mail on this domain.)" % domain),
	]
	for qname, rtype, value, explanation in defaults:
		if value is None or value.strip() == "": continue # skip IPV6 if not set
		if not is_zone and qname == "www": continue # don't create any default 'www' subdomains on what are themselves subdomains
		# Set the default record, but not if:
		# (1) there is not a user-set record of the same type already
		# (2) there is not a CNAME record already, since you can't set both and who knows what takes precedence
		# (2) there is not an A record already (if this is an A record this is a dup of (1), and if this is an AAAA record then don't set a default AAAA record if the user sets a custom A record, since the default wouldn't make sense and it should not resolve if the user doesn't provide a new AAAA record)
		if not has_rec(qname, rtype) and not has_rec(qname, "CNAME") and not has_rec(qname, "A"):
			records.append((qname, rtype, value, explanation))

	# Don't pin the list of records that has_rec checks against anymore.
	has_rec_base = records

	if domain_properties[domain]["mail"]:
		# The MX record says where email for the domain should be delivered: Here!
		if not has_rec(None, "MX", prefix="10 "):
			records.append((None,  "MX",  "10 %s." % env["PRIMARY_HOSTNAME"], "Required. Specifies the hostname (and priority) of the machine that handles @%s mail." % domain))

		# SPF record: Permit the box ('mx', see above) to send mail on behalf of
		# the domain, and no one else.
		# Skip if the user has set a custom SPF record.
		if not has_rec(None, "TXT", prefix="v=spf1 "):
			records.append((None,  "TXT", 'v=spf1 mx -all', "Recommended. Specifies that only the box is permitted to send @%s mail." % domain))

		# Append the DKIM TXT record to the zone as generated by OpenDKIM.
		# Skip if the user has set a DKIM record already.
		opendkim_record_file = os.path.join(env['STORAGE_ROOT'], 'mail/dkim/mail.txt')
		with open(opendkim_record_file, encoding="utf-8") as orf:
			m = re.match(r'(\S+)\s+IN\s+TXT\s+\( ((?:"[^"]+"\s+)+)\)', orf.read(), re.S)
			val = "".join(re.findall(r'"([^"]+)"', m.group(2)))
			if not has_rec(m.group(1), "TXT", prefix="v=DKIM1; "):
				records.append((m.group(1), "TXT", val, "Recommended. Provides a way for recipients to verify that this machine sent @%s mail." % domain))

		# Append a DMARC record.
		# Skip if the user has set a DMARC record already.
		if not has_rec("_dmarc", "TXT", prefix="v=DMARC1; "):
			records.append(("_dmarc", "TXT", 'v=DMARC1; p=quarantine;', "Recommended. Specifies that mail that does not originate from the box but claims to be from @%s or which does not have a valid DKIM signature is suspect and should be quarantined by the recipient's mail system." % domain))

	if domain_properties[domain]["user"]:
		# Add CardDAV/CalDAV SRV records on the non-primary hostname that points to the primary hostname
		# for autoconfiguration of mail clients (so only domains hosting user accounts need it).
		# The SRV record format is priority (0, whatever), weight (0, whatever), port, service provider hostname (w/ trailing dot).
		if domain != env["PRIMARY_HOSTNAME"]:
			for dav in ("card", "cal"):
				qname = "_" + dav + "davs._tcp"
				if not has_rec(qname, "SRV"):
					records.append((qname, "SRV", "0 0 443 " + env["PRIMARY_HOSTNAME"] + ".", "Recommended. Specifies the hostname of the server that handles CardDAV/CalDAV services for email addresses on this domain."))

	# If this is a domain name that there are email addresses configured for, i.e. "something@"
	# this domain name, then the domain name is a MTA-STS (https://tools.ietf.org/html/rfc8461)
	# Policy Domain.
	#
	# A "_mta-sts" TXT record signals the presence of a MTA-STS policy. The id field helps clients
	# cache the policy. It should be stable so we don't update DNS unnecessarily but change when
	# the policy changes. It must be at most 32 letters and numbers, so we compute a hash of the
	# policy file.
	#
	# The policy itself is served at the "mta-sts" (no underscore) subdomain over HTTPS. Therefore
	# the TLS certificate used by Postfix for STARTTLS must be a valid certificate for the MX
	# domain name (PRIMARY_HOSTNAME) *and* the TLS certificate used by nginx for HTTPS on the mta-sts
	# subdomain must be valid certificate for that domain. Do not set an MTA-STS policy if either
	# certificate in use is not valid (e.g. because it is self-signed and a valid certificate has not
	# yet been provisioned). Since we cannot provision a certificate without A/AAAA records, we
	# always set them (by including them in the www domains) --- only the TXT records depend on there
	# being valid certificates.
	mta_sts_records = [ ]
	if domain_properties[domain]["mail"] \
	  and domain_properties[env["PRIMARY_HOSTNAME"]]["certificate-is-valid"] \
	  and is_domain_cert_signed_and_valid("mta-sts." + domain, env):
		# Compute an up-to-32-character hash of the policy file. We'll take a SHA-1 hash of the policy
		# file (20 bytes) and encode it as base-64 (28 bytes, using alphanumeric alternate characters
		# instead of '+' and '/' which are not allowed in an MTA-STS policy id) but then just take its
		# first 20 characters, which is more than sufficient to change whenever the policy file changes
		# (and ensures any '=' padding at the end of the base64 encoding is dropped).
		with open("/var/lib/mailinabox/mta-sts.txt", "rb") as f:
			mta_sts_policy_id = base64.b64encode(hashlib.sha1(f.read()).digest(), altchars=b"AA").decode("ascii")[0:20]
		mta_sts_records.extend([
			("_mta-sts", "TXT", "v=STSv1; id=" + mta_sts_policy_id, "Optional. Part of the MTA-STS policy for incoming mail. If set, a MTA-STS policy must also be published.")
		])

		# Enable SMTP TLS reporting (https://tools.ietf.org/html/rfc8460) if the user has set a config option.
		# Skip if the rules below if the user has set a custom _smtp._tls record.
		if env.get("MTA_STS_TLSRPT_RUA") and not has_rec("_smtp._tls", "TXT", prefix="v=TLSRPTv1;"):
			mta_sts_records.append(("_smtp._tls", "TXT", "v=TLSRPTv1; rua=" + env["MTA_STS_TLSRPT_RUA"], "Optional. Enables MTA-STS reporting."))
	for qname, rtype, value, explanation in mta_sts_records:
		if not has_rec(qname, rtype):
			records.append((qname, rtype, value, explanation))

	# Add no-mail-here records for any qname that has an A or AAAA record
	# but no MX record. This would include domain itself if domain is a
	# non-mail domain and also may include qnames from custom DNS records.
	# Do this once at the end of generating a zone.
	if is_zone:
		qnames_with_a = {qname for (qname, rtype, value, explanation) in records if rtype in {"A", "AAAA"}}
		qnames_with_mx = {qname for (qname, rtype, value, explanation) in records if rtype == "MX"}
		for qname in qnames_with_a - qnames_with_mx:
			# Mark this domain as not sending mail with hard-fail SPF and DMARC records.
			d = (qname+"." if qname else "") + domain
			if not has_rec(qname, "TXT", prefix="v=spf1 "):
				records.append((qname,  "TXT", 'v=spf1 -all', "Recommended. Prevents use of this domain name for outbound mail by specifying that no servers are valid sources for mail from @%s. If you do send email from this domain name you should either override this record such that the SPF rule does allow the originating server, or, take the recommended approach and have the box handle mail for this domain (simply add any receiving alias at this domain name to make this machine treat the domain name as one of its mail domains)." % d))
			if not has_rec("_dmarc" + ("."+qname if qname else ""), "TXT", prefix="v=DMARC1; "):
				records.append(("_dmarc" + ("."+qname if qname else ""), "TXT", 'v=DMARC1; p=reject;', "Recommended. Prevents use of this domain name for outbound mail by specifying that the SPF rule should be honoured for mail from @%s." % d))

			# And with a null MX record (https://explained-from-first-principles.com/email/#null-mx-record)
			if not has_rec(qname, "MX"):
				records.append((qname, "MX", '0 .', "Recommended. Prevents use of this domain name for incoming mail."))

	# Sort the records. The None records *must* go first in the nsd zone file. Otherwise it doesn't matter.
	records.sort(key = lambda rec : list(reversed(rec[0].split(".")) if rec[0] is not None else ""))

	return records

def is_domain_cert_signed_and_valid(domain, env):
	cert = get_ssl_certificates(env).get(domain)
	if not cert: return False # no certificate provisioned
	cert_status = check_certificate(domain, cert['certificate'], cert['private-key'])
	return cert_status[0] == 'OK'

########################################################################

def build_tlsa_record(env):
	# A DANE TLSA record in DNS specifies that connections on a port
	# must use TLS and the certificate must match a particular criteria.
	#
	# Thanks to http://blog.huque.com/2012/10/dnssec-and-certificates.html
	# and https://community.letsencrypt.org/t/please-avoid-3-0-1-and-3-0-2-dane-tlsa-records-with-le-certificates/7022
	# for explaining all of this! Also see https://tools.ietf.org/html/rfc6698#section-2.1
	# and https://github.com/mail-in-a-box/mailinabox/issues/268#issuecomment-167160243.
	#
	# There are several criteria. We used to use "3 0 1" criteria, which
	# meant to pin a leaf (3) certificate (0) with SHA256 hash (1). But
	# certificates change, and especially as we move to short-lived certs
	# they change often. The TLSA record handily supports the criteria of
	# a leaf certificate (3)'s subject public key (1) with SHA256 hash (1).
	# The subject public key is the public key portion of the private key
	# that generated the CSR that generated the certificate. Since we
	# generate a private key once the first time Mail-in-a-Box is set up
	# and reuse it for all subsequent certificates, the TLSA record will
	# remain valid indefinitely.

	from ssl_certificates import load_cert_chain, load_pem
	from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat

	fn = os.path.join(env["STORAGE_ROOT"], "ssl", "ssl_certificate.pem")
	cert = load_pem(load_cert_chain(fn)[0])

	subject_public_key = cert.public_key().public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)
	# We could have also loaded ssl_private_key.pem and called priv_key.public_key().public_bytes(...)

	pk_hash = hashlib.sha256(subject_public_key).hexdigest()

	# Specify the TLSA parameters:
	# 3: Match the (leaf) certificate. (No CA, no trust path needed.)
	# 1: Match its subject public key.
	# 1: Use SHA256.
	return "3 1 1 " + pk_hash

def build_sshfp_records():
	# The SSHFP record is a way for us to embed this server's SSH public
	# key fingerprint into the DNS so that remote hosts have an out-of-band
	# method to confirm the fingerprint. See RFC 4255 and RFC 6594. This
	# depends on DNSSEC.
	#
	# On the client side, set SSH's VerifyHostKeyDNS option to 'ask' to
	# include this info in the key verification prompt or 'yes' to trust
	# the SSHFP record.
	#
	# See https://github.com/xelerance/sshfp for inspiriation.

	algorithm_number = {
		"ssh-rsa": 1,
		"ssh-dss": 2,
		"ecdsa-sha2-nistp256": 3,
		"ssh-ed25519": 4,
	}

	# Get our local fingerprints by running ssh-keyscan. The output looks
	# like the known_hosts file: hostname, keytype, fingerprint. The order
	# of the output is arbitrary, so sort it to prevent spurious updates
	# to the zone file (that trigger bumping the serial number). However,
	# if SSH has been configured to listen on a nonstandard port, we must
	# specify that port to sshkeyscan.

	port = get_ssh_port()

	# If nothing returned, SSH is probably not installed.
	if not port:
		return

	keys = shell("check_output", ["ssh-keyscan", "-4", "-t", "rsa,dsa,ecdsa,ed25519", "-p", str(port), "localhost"])
	keys = sorted(keys.split("\n"))

	for key in keys:
		if key.strip() == "" or key[0] == "#": continue
		try:
			_host, keytype, pubkey = key.split(" ")
			yield "%d %d ( %s )" % (
				algorithm_number[keytype],
				2, # specifies we are using SHA-256 on next line
				hashlib.sha256(base64.b64decode(pubkey)).hexdigest().upper(),
				)
		except:
			# Lots of things can go wrong. Don't let it disturb the DNS
			# zone.
			pass

########################################################################

def write_nsd_zone(domain, zonefile, records, env, force):
	# On the $ORIGIN line, there's typically a ';' comment at the end explaining
	# what the $ORIGIN line does. Any further data after the domain confuses
	# ldns-signzone, however. It used to say '; default zone domain'.
	#
	# The SOA contact address for all of the domains on this system is hostmaster
	# @ the PRIMARY_HOSTNAME. Hopefully that's legit.
	#
	# For the refresh through TTL fields, a good reference is:
	# https://www.ripe.net/publications/docs/ripe-203
	#
	# A hash of the available DNSSEC keys are added in a comment so that when
	# the keys change we force a re-generation of the zone which triggers
	# re-signing it.

	zone = """
$ORIGIN {domain}.
$TTL 86400          ; default time to live

@ IN SOA ns1.{primary_domain}. hostmaster.{primary_domain}. (
           __SERIAL__     ; serial number
           7200     ; Refresh (secondary nameserver update interval)
           3600     ; Retry (when refresh fails, how often to try again, should be lower than the refresh)
           1209600  ; Expire (when refresh fails, how long secondary nameserver will keep records around anyway)
           86400    ; Negative TTL (how long negative responses are cached)
           )
"""

	# Replace replacement strings.
	zone = zone.format(domain=domain, primary_domain=env["PRIMARY_HOSTNAME"])

	# Add records.
	for subdomain, querytype, value, _explanation in records:
		if subdomain:
			zone += subdomain
		zone += "\tIN\t" + querytype + "\t"
		if querytype == "TXT":
			# Divide into 255-byte max substrings.
			v2 = ""
			while len(value) > 0:
				s = value[0:255]
				value = value[255:]
				s = s.replace('\\', '\\\\') # escape backslashes
				s = s.replace('"', '\\"') # escape quotes
				s = '"' + s + '"' # wrap in quotes
				v2 += s + " "
			value = v2
		zone += value + "\n"

	# Append a stable hash of DNSSEC signing keys in a comment.
	zone += f"\n; DNSSEC signing keys hash: {hash_dnssec_keys(domain, env)}\n"

	# DNSSEC requires re-signing a zone periodically. That requires
	# bumping the serial number even if no other records have changed.
	# We don't see the DNSSEC records yet, so we have to figure out
	# if a re-signing is necessary so we can prematurely bump the
	# serial number.
	force_bump = False
	if not os.path.exists(zonefile + ".signed"):
		# No signed file yet. Shouldn't normally happen unless a box
		# is going from not using DNSSEC to using DNSSEC.
		force_bump = True
	else:
		# We've signed the domain. Check if we are close to the expiration
		# time of the signature. If so, we'll force a bump of the serial
		# number so we can re-sign it.
		with open(zonefile + ".signed", encoding="utf-8") as f:
			signed_zone = f.read()
		expiration_times = re.findall(r"\sRRSIG\s+SOA\s+\d+\s+\d+\s\d+\s+(\d{14})", signed_zone)
		if len(expiration_times) == 0:
			# weird
			force_bump = True
		else:
			# All of the times should be the same, but if not choose the soonest.
			expiration_time = min(expiration_times)
			expiration_time = datetime.datetime.strptime(expiration_time, "%Y%m%d%H%M%S")
			if expiration_time - datetime.datetime.now() < datetime.timedelta(days=3):
				# We're within three days of the expiration, so bump serial & resign.
				force_bump = True

	# Set the serial number.
	serial = datetime.datetime.now().strftime("%Y%m%d00")
	if os.path.exists(zonefile):
		# If the zone already exists, is different, and has a later serial number,
		# increment the number.
		with open(zonefile, encoding="utf-8") as f:
			existing_zone = f.read()
			m = re.search(r"(\d+)\s*;\s*serial number", existing_zone)
			if m:
				# Clear out the serial number in the existing zone file for the
				# purposes of seeing if anything *else* in the zone has changed.
				existing_serial = m.group(1)
				existing_zone = existing_zone.replace(m.group(0), "__SERIAL__     ; serial number")

				# If the existing zone is the same as the new zone (modulo the serial number),
				# there is no need to update the file. Unless we're forcing a bump.
				if zone == existing_zone and not force_bump and not force:
					return False

				# If the existing serial is not less than a serial number
				# based on the current date plus 00, increment it. Otherwise,
				# the serial number is less than our desired new serial number
				# so we'll use the desired new number.
				if existing_serial >= serial:
					serial = str(int(existing_serial) + 1)

	zone = zone.replace("__SERIAL__", serial)

	# Write the zone file.
	with open(zonefile, "w", encoding="utf-8") as f:
		f.write(zone)

	return True # file is updated

def get_dns_zonefile(zone, env):
	for domain, fn in get_dns_zones(env):
		if zone == domain:
			break
	else:
		raise ValueError("%s is not a domain name that corresponds to a zone." % zone)

	nsd_zonefile = "/etc/nsd/zones/" + fn
	with open(nsd_zonefile, encoding="utf-8") as f:
		return f.read()

########################################################################

def write_nsd_conf(zonefiles, additional_records, env):
	# Write the list of zones to a configuration file.
	nsd_conf_file = "/etc/nsd/nsd.conf.d/zones.conf"
	nsdconf = ""

	# Append the zones.
	for domain, zonefile in zonefiles:
		nsdconf += f"""
zone:
	name: {domain}
	zonefile: {zonefile}
"""

		# If custom secondary nameservers have been set, allow zone transfers
		# and, if not a subnet, notifies to them.
		for ipaddr in get_secondary_dns(additional_records, mode="xfr"):
			if "/" not in ipaddr:
				nsdconf += "\n\tnotify: %s NOKEY" % (ipaddr)
			nsdconf += "\n\tprovide-xfr: %s NOKEY\n" % (ipaddr)

	# Check if the file is changing. If it isn't changing,
	# return False to flag that no change was made.
	if os.path.exists(nsd_conf_file):
		with open(nsd_conf_file, encoding="utf-8") as f:
			if f.read() == nsdconf:
				return False

	# Write out new contents and return True to signal that
	# configuration changed.
	with open(nsd_conf_file, "w", encoding="utf-8") as f:
		f.write(nsdconf)
	return True

########################################################################

def find_dnssec_signing_keys(domain, env):
	# For key that we generated (one per algorithm)...
	d = os.path.join(env['STORAGE_ROOT'], 'dns/dnssec')
	keyconfs = [f for f in os.listdir(d) if f.endswith(".conf")]
	for keyconf in keyconfs:
		# Load the file holding the KSK and ZSK key filenames.
		keyconf_fn = os.path.join(d, keyconf)
		keyinfo = load_env_vars_from_file(keyconf_fn)

		# Skip this key if the conf file has a setting named DOMAINS,
		# holding a comma-separated list of domain names, and if this
		# domain is not in the list. This allows easily disabling a
		# key by setting "DOMAINS=" or "DOMAINS=none", other than
		# deleting the key's .conf file, which might result in the key
		# being regenerated next upgrade. Keys should be disabled if
		# they are not needed to reduce the DNSSEC query response size.
		if "DOMAINS" in keyinfo and domain not in [dd.strip() for dd in keyinfo["DOMAINS"].split(",")]:
			continue

		for keytype in ("KSK", "ZSK"):
			yield keytype, keyinfo[keytype]

def hash_dnssec_keys(domain, env):
	# Create a stable (by sorting the items) hash of all of the private keys
	# that will be used to sign this domain.
	keydata = []
	for keytype, keyfn in sorted(find_dnssec_signing_keys(domain, env)):
		oldkeyfn = os.path.join(env['STORAGE_ROOT'], 'dns/dnssec', keyfn + ".private")
		keydata.extend((keytype, keyfn))
		with open(oldkeyfn, encoding="utf-8") as fr:
			keydata.append( fr.read() )
	keydata = "".join(keydata).encode("utf8")
	return hashlib.sha1(keydata).hexdigest()

def sign_zone(domain, zonefile, env):
	# Sign the zone with all of the keys that were generated during
	# setup so that the user can choose which to use in their DS record at
	# their registrar, and also to support migration to newer algorithms.

	# In order to use the key files generated at setup which are for
	# the domain _domain_, we have to re-write the files and place
	# the actual domain name in it, so that ldns-signzone works.
	#
	# Patch each key, storing the patched version in /tmp for now.
	# Each key has a .key and .private file. Collect a list of filenames
	# for all of the keys (and separately just the key-signing keys).
	all_keys = []
	ksk_keys = []
	for keytype, keyfn in find_dnssec_signing_keys(domain, env):
		newkeyfn = '/tmp/' + keyfn.replace("_domain_", domain)

		for ext in (".private", ".key"):
			# Copy the .key and .private files to /tmp to patch them up.
			#
			# Use os.umask and open().write() to securely create a copy that only
			# we (root) can read.
			oldkeyfn = os.path.join(env['STORAGE_ROOT'], 'dns/dnssec', keyfn + ext)
			with open(oldkeyfn, encoding="utf-8") as fr:
				keydata = fr.read()
			keydata = keydata.replace("_domain_", domain)
			prev_umask = os.umask(0o77) # ensure written file is not world-readable
			try:
				with open(newkeyfn + ext, "w", encoding="utf-8") as fw:
					fw.write(keydata)
			finally:
				os.umask(prev_umask) # other files we write should be world-readable

		# Put the patched key filename base (without extension) into the list of keys we'll sign with.
		all_keys.append(newkeyfn)
		if keytype == "KSK": ksk_keys.append(newkeyfn)

	# Do the signing.
	expiry_date = (datetime.datetime.now() + datetime.timedelta(days=30)).strftime("%Y%m%d")
	shell('check_call', ["/usr/bin/ldns-signzone",
		# expire the zone after 30 days
		"-e", expiry_date,

		# use NSEC3
		"-n",

		# zonefile to sign
		"/etc/nsd/zones/" + zonefile,
	]
		# keys to sign with (order doesn't matter -- it'll figure it out)
		+ all_keys
	)

	# Create a DS record based on the patched-up key files. The DS record is specific to the
	# zone being signed, so we can't use the .ds files generated when we created the keys.
	# The DS record points to the KSK only. Write this next to the zone file so we can
	# get it later to give to the user with instructions on what to do with it.
	#
	# Generate a DS record for each key. There are also several possible hash algorithms that may
	# be used, so we'll pre-generate all for each key. One DS record per line. Only one
	# needs to actually be deployed at the registrar. We'll select the preferred one
	# in the status checks.
	with open("/etc/nsd/zones/" + zonefile + ".ds", "w", encoding="utf-8") as f:
		for key in ksk_keys:
			for digest_type in ('1', '2', '4'):
				rr_ds = shell('check_output', ["/usr/bin/ldns-key2ds",
					"-n", # output to stdout
					"-" + digest_type, # 1=SHA1, 2=SHA256, 4=SHA384
					key + ".key"
				])
				f.write(rr_ds)

	# Remove the temporary patched key files.
	for fn in all_keys:
		os.unlink(fn + ".private")
		os.unlink(fn + ".key")

########################################################################

def write_opendkim_tables(domains, env):
	# Append a record to OpenDKIM's KeyTable and SigningTable for each domain
	# that we send mail from (zones and all subdomains).

	opendkim_key_file = os.path.join(env['STORAGE_ROOT'], 'mail/dkim/mail.private')

	if not os.path.exists(opendkim_key_file):
		# Looks like OpenDKIM is not installed.
		return False

	config = {
		# The SigningTable maps email addresses to a key in the KeyTable that
		# specifies signing information for matching email addresses. Here we
		# map each domain to a same-named key.
		#
		# Elsewhere we set the DMARC policy for each domain such that mail claiming
		# to be From: the domain must be signed with a DKIM key on the same domain.
		# So we must have a separate KeyTable entry for each domain.
		"SigningTable":
			"".join(
				f"*@{domain} {domain}\n"
				for domain in domains
			),

		# The KeyTable specifies the signing domain, the DKIM selector, and the
		# path to the private key to use for signing some mail. Per DMARC, the
		# signing domain must match the sender's From: domain.
		"KeyTable":
			"".join(
				f"{domain} {domain}:mail:{opendkim_key_file}\n"
				for domain in domains
			),
	}

	did_update = False
	for filename, content in config.items():
		# Don't write the file if it doesn't need an update.
		if os.path.exists("/etc/opendkim/" + filename):
			with open("/etc/opendkim/" + filename, encoding="utf-8") as f:
				if f.read() == content:
					continue

		# The contents needs to change.
		with open("/etc/opendkim/" + filename, "w", encoding="utf-8") as f:
			f.write(content)
		did_update = True

	# Return whether the files changed. If they didn't change, there's
	# no need to kick the opendkim process.
	return did_update

########################################################################

def get_custom_dns_config(env, only_real_records=False):
	try:
		with open(os.path.join(env['STORAGE_ROOT'], 'dns/custom.yaml'), encoding="utf-8") as f:
			custom_dns = rtyaml.load(f)
		if not isinstance(custom_dns, dict): raise ValueError # caught below
	except:
		return [ ]

	for qname, value in custom_dns.items():
		if qname == "_secondary_nameserver" and only_real_records: continue # skip fake record

		# Short form. Mapping a domain name to a string is short-hand
		# for creating A records.
		if isinstance(value, str):
			values = [("A", value)]

		# A mapping creates multiple records.
		elif isinstance(value, dict):
			values = value.items()

		# No other type of data is allowed.
		else:
			raise ValueError

		for rtype, value2 in values:
			if isinstance(value2, str):
				yield (qname, rtype, value2)
			elif isinstance(value2, list):
				for value3 in value2:
					yield (qname, rtype, value3)
			# No other type of data is allowed.
			else:
				raise ValueError

def filter_custom_records(domain, custom_dns_iter):
	for qname, rtype, value in custom_dns_iter:
		# We don't count the secondary nameserver config (if present) as a record - that would just be
		# confusing to users. Instead it is accessed/manipulated directly via (get/set)_custom_dns_config.
		if qname == "_secondary_nameserver": continue

		# Is this record for the domain or one of its subdomains?
		# If `domain` is None, return records for all domains.
		if domain is not None and qname != domain and not qname.endswith("." + domain): continue

		# Turn the fully qualified domain name in the YAML file into
		# our short form (None => domain, or a relative QNAME) if
		# domain is not None.
		if domain is not None:
			qname = None if qname == domain else qname[0:len(qname) - len("." + domain)]

		yield (qname, rtype, value)

def write_custom_dns_config(config, env):
	# We get a list of (qname, rtype, value) triples. Convert this into a
	# nice dictionary format for storage on disk.
	from collections import OrderedDict
	config = list(config)
	dns = OrderedDict()
	seen_qnames = set()

	# Process the qnames in the order we see them.
	for qname in [rec[0] for rec in config]:
		if qname in seen_qnames: continue
		seen_qnames.add(qname)

		records = [(rec[1], rec[2]) for rec in config if rec[0] == qname]
		if len(records) == 1 and records[0][0] == "A":
			dns[qname] = records[0][1]
		else:
			dns[qname] = OrderedDict()
			seen_rtypes = set()

			# Process the rtypes in the order we see them.
			for rtype in [rec[0] for rec in records]:
				if rtype in seen_rtypes: continue
				seen_rtypes.add(rtype)

				values = [rec[1] for rec in records if rec[0] == rtype]
				if len(values) == 1:
					values = values[0]
				dns[qname][rtype] = values

	# Write.
	config_yaml = rtyaml.dump(dns)
	with open(os.path.join(env['STORAGE_ROOT'], 'dns/custom.yaml'), "w", encoding="utf-8") as f:
		f.write(config_yaml)

def set_custom_dns_record(qname, rtype, value, action, env):
	# validate qname
	for zone, _fn in get_dns_zones(env):
		# It must match a zone apex or be a subdomain of a zone
		# that we are otherwise hosting.
		if qname == zone or qname.endswith("."+zone):
			break
	else:
		# No match.
		if qname != "_secondary_nameserver":
			raise ValueError("%s is not a domain name or a subdomain of a domain name managed by this box." % qname)

	# validate rtype
	rtype = rtype.upper()
	if value is not None and qname != "_secondary_nameserver":
		if not re.search(DOMAIN_RE, qname):
			msg = "Invalid name."
			raise ValueError(msg)

		if rtype in {"A", "AAAA"}:
			if value != "local": # "local" is a special flag for us
				v = ipaddress.ip_address(value) # raises a ValueError if there's a problem
				if rtype == "A" and not isinstance(v, ipaddress.IPv4Address): raise ValueError("That's an IPv6 address.")
				if rtype == "AAAA" and not isinstance(v, ipaddress.IPv6Address): raise ValueError("That's an IPv4 address.")
		elif rtype in {"CNAME", "NS"}:
			if rtype == "NS" and qname == zone:
				msg = "NS records can only be set for subdomains."
				raise ValueError(msg)

			# ensure value has a trailing dot
			if not value.endswith("."):
				value = value + "."

			if not re.search(DOMAIN_RE, value):
				msg = "Invalid value."
				raise ValueError(msg)
		elif rtype in {"CNAME", "TXT", "SRV", "MX", "SSHFP", "CAA"}:
			# anything goes
			pass
		else:
			raise ValueError("Unknown record type '%s'." % rtype)

	# load existing config
	config = list(get_custom_dns_config(env))

	# update
	newconfig = []
	made_change = False
	needs_add = True
	for _qname, _rtype, _value in config:
		if action == "add":
			if (_qname, _rtype, _value) == (qname, rtype, value):
				# Record already exists. Bail.
				return False
		elif action == "set":
			if (_qname, _rtype) == (qname, rtype):
				if _value == value:
					# Flag that the record already exists, don't
					# need to add it.
					needs_add = False
				else:
					# Drop any other values for this (qname, rtype).
					made_change = True
					continue
		elif action == "remove":
			if (_qname, _rtype, _value) == (qname, rtype, value):
				# Drop this record.
				made_change = True
				continue
			if value is None and (_qname, _rtype) == (qname, rtype):
				# Drop all qname-rtype records.
				made_change = True
				continue
		else:
			raise ValueError("Invalid action: " + action)

		# Preserve this record.
		newconfig.append((_qname, _rtype, _value))

	if action in {"add", "set"} and needs_add and value is not None:
		newconfig.append((qname, rtype, value))
		made_change = True

	if made_change:
		# serialize & save
		write_custom_dns_config(newconfig, env)
	return made_change

########################################################################

def get_secondary_dns(custom_dns, mode=None):
	resolver = dns.resolver.get_default_resolver()
	resolver.timeout = 10
	resolver.lifetime = 10

	values = []
	for qname, _rtype, value in custom_dns:
		if qname != '_secondary_nameserver': continue
		for hostname in value.split(" "):
			hostname = hostname.strip()
			if mode is None:
				# Just return the setting.
				values.append(hostname)
				continue

			# If the entry starts with "xfr:" only include it in the zone transfer settings.
			if hostname.startswith("xfr:"):
				if mode != "xfr": continue
				hostname = hostname[4:]

			# If is a hostname, before including in zone xfr lines,
			# resolve to an IP address.
			# It may not resolve to IPv6, so don't throw an exception if it
			# doesn't. Skip the entry if there is a DNS error.
			if mode == "xfr":
				try:
					ipaddress.ip_interface(hostname) # test if it's an IP address or CIDR notation
					values.append(hostname)
				except ValueError:
					try:
						response = dns.resolver.resolve(hostname+'.', "A", raise_on_no_answer=False)
						values.extend(map(str, response))
					except dns.exception.DNSException:
						pass
					try:
						response = dns.resolver.resolve(hostname+'.', "AAAA", raise_on_no_answer=False)
						values.extend(map(str, response))
					except dns.exception.DNSException:
						pass

			else:
				values.append(hostname)

	return values

def set_secondary_dns(hostnames, env):
	if len(hostnames) > 0:
		# Validate that all hostnames are valid and that all zone-xfer IP addresses are valid.
		resolver = dns.resolver.get_default_resolver()
		resolver.timeout = 5
		resolver.lifetime = 5

		for item in hostnames:
			if not item.startswith("xfr:"):
				# Resolve hostname.
				try:
					resolver.resolve(item, "A")
				except (dns.resolver.NoNameservers, dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.resolver.Timeout):
					try:
						resolver.resolve(item, "AAAA")
					except (dns.resolver.NoNameservers, dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.resolver.Timeout):
						raise ValueError("Could not resolve the IP address of %s." % item)
			else:
				# Validate IP address.
				try:
					if "/" in item[4:]:
						ipaddress.ip_network(item[4:]) # raises a ValueError if there's a problem
					else:
						ipaddress.ip_address(item[4:]) # raises a ValueError if there's a problem
				except ValueError:
					raise ValueError("'%s' is not an IPv4 or IPv6 address or subnet." % item[4:])

		# Set.
		set_custom_dns_record("_secondary_nameserver", "A", " ".join(hostnames), "set", env)
	else:
		# Clear.
		set_custom_dns_record("_secondary_nameserver", "A", None, "set", env)

	# Apply.
	return do_dns_update(env)


def get_custom_dns_records(custom_dns, qname, rtype):
	for qname1, rtype1, value in custom_dns:
		if qname1 == qname and rtype1 == rtype:
			yield value

########################################################################

def build_recommended_dns(env):
	ret = []
	for (domain, _zonefile, records) in build_zones(env):
		# remove records that we don't display
		records = [r for r in records if r[3] is not False]

		# put Required at the top, then Recommended, then everythiing else
		records.sort(key = lambda r : 0 if r[3].startswith("Required.") else (1 if r[3].startswith("Recommended.") else 2))

		# expand qnames
		for i in range(len(records)):
			qname = domain if records[i][0] is None else records[i][0] + "." + domain

			records[i] = {
				"qname": qname,
				"rtype": records[i][1],
				"value": records[i][2],
				"explanation": records[i][3],
			}

		# return
		ret.append((domain, records))
	return ret

if __name__ == "__main__":
	from utils import load_environment
	env = load_environment()
	if sys.argv[-1] == "--lint":
		write_custom_dns_config(get_custom_dns_config(env), env)
	else:
		for _zone, records in build_recommended_dns(env):
			for record in records:
				print("; " + record['explanation'])
				print(record['qname'], record['rtype'], record['value'], sep="\t")
				print()



File: management/mail_log.py

#!/usr/local/lib/mailinabox/env/bin/python
import argparse
import datetime
import gzip
import os.path
import re
import shutil
import tempfile
import textwrap
from collections import defaultdict, OrderedDict

import dateutil.parser
import time

from dateutil.relativedelta import relativedelta

import utils


LOG_FILES = (
    '/var/log/mail.log.6.gz',
    '/var/log/mail.log.5.gz',
    '/var/log/mail.log.4.gz',
    '/var/log/mail.log.3.gz',
    '/var/log/mail.log.2.gz',
    '/var/log/mail.log.1',
    '/var/log/mail.log',
)

TIME_DELTAS = OrderedDict([
    ('all', datetime.timedelta(weeks=52)),
    ('month', datetime.timedelta(weeks=4)),
    ('2weeks', datetime.timedelta(days=14)),
    ('week', datetime.timedelta(days=7)),
    ('2days', datetime.timedelta(days=2)),
    ('day', datetime.timedelta(days=1)),
    ('12hours', datetime.timedelta(hours=12)),
    ('6hours', datetime.timedelta(hours=6)),
    ('hour', datetime.timedelta(hours=1)),
    ('30min', datetime.timedelta(minutes=30)),
    ('10min', datetime.timedelta(minutes=10)),
    ('5min', datetime.timedelta(minutes=5)),
    ('min', datetime.timedelta(minutes=1)),
    ('today', datetime.datetime.now() - datetime.datetime.now().replace(hour=0, minute=0, second=0))
])

END_DATE = NOW = datetime.datetime.now()
START_DATE = None

VERBOSE = False

# List of strings to filter users with
FILTERS = None

# What to show (with defaults)
SCAN_OUT = True  # Outgoing email
SCAN_IN = True  # Incoming email
SCAN_DOVECOT_LOGIN = True  # Dovecot Logins
SCAN_GREY = False  # Greylisted email
SCAN_BLOCKED = False  # Rejected email


def scan_files(collector):
    """ Scan files until they run out or the earliest date is reached """

    stop_scan = False

    for fn in LOG_FILES:

        tmp_file = None

        if not os.path.exists(fn):
            continue
        elif fn[-3:] == '.gz':
            tmp_file = tempfile.NamedTemporaryFile()
            with gzip.open(fn, 'rb') as f:
                shutil.copyfileobj(f, tmp_file)

        if VERBOSE:
            print("Processing file", fn, "...")
        fn = tmp_file.name if tmp_file else fn

        for line in readline(fn):
            if scan_mail_log_line(line.strip(), collector) is False:
                if stop_scan:
                    return
                stop_scan = True
            else:
                stop_scan = False



def scan_mail_log(env):
    """ Scan the system's mail log files and collect interesting data

    This function scans the 2 most recent mail log files in /var/log/.

    Args:
        env (dict): Dictionary containing MiaB settings

    """

    collector = {
        "scan_count": 0,  # Number of lines scanned
        "parse_count": 0,  # Number of lines parsed (i.e. that had their contents examined)
        "scan_time": time.time(),  # The time in seconds the scan took
        "sent_mail": OrderedDict(),  # Data about email sent by users
        "received_mail": OrderedDict(),  # Data about email received by users
        "logins": OrderedDict(),  # Data about login activity
        "postgrey": {},  # Data about greylisting of email addresses
        "rejected": OrderedDict(),  # Emails that were blocked
        "known_addresses": None,  # Addresses handled by the Miab installation
        "other-services": set(),
    }

    try:
        import mailconfig
        collector["known_addresses"] = (set(mailconfig.get_mail_users(env)) |
                                        {alias[0] for alias in mailconfig.get_mail_aliases(env)})
    except ImportError:
        pass

    print(f"Scanning logs from {START_DATE:%Y-%m-%d %H:%M:%S} to {END_DATE:%Y-%m-%d %H:%M:%S}"
    )

    # Scan the lines in the log files until the date goes out of range
    scan_files(collector)

    if not collector["scan_count"]:
        print("No log lines scanned...")
        return

    collector["scan_time"] = time.time() - collector["scan_time"]

    print("{scan_count} Log lines scanned, {parse_count} lines parsed in {scan_time:.2f} "
          "seconds\n".format(**collector))

    # Print Sent Mail report

    if collector["sent_mail"]:
        msg = "Sent email"
        print_header(msg)

        data = OrderedDict(sorted(collector["sent_mail"].items(), key=email_sort))

        print_user_table(
            data.keys(),
            data=[
                ("sent", [u["sent_count"] for u in data.values()]),
                ("hosts", [len(u["hosts"]) for u in data.values()]),
            ],
            sub_data=[
                ("sending hosts", [u["hosts"] for u in data.values()]),
            ],
            activity=[
                ("sent", [u["activity-by-hour"] for u in data.values()]),
            ],
            earliest=[u["earliest"] for u in data.values()],
            latest=[u["latest"] for u in data.values()],
        )

        accum = defaultdict(int)
        data = collector["sent_mail"].values()

        for h in range(24):
            accum[h] = sum(d["activity-by-hour"][h] for d in data)

        print_time_table(
            ["sent"],
            [accum]
        )

    # Print Received Mail report

    if collector["received_mail"]:
        msg = "Received email"
        print_header(msg)

        data = OrderedDict(sorted(collector["received_mail"].items(), key=email_sort))

        print_user_table(
            data.keys(),
            data=[
                ("received", [u["received_count"] for u in data.values()]),
            ],
            activity=[
                ("sent", [u["activity-by-hour"] for u in data.values()]),
            ],
            earliest=[u["earliest"] for u in data.values()],
            latest=[u["latest"] for u in data.values()],
        )

        accum = defaultdict(int)
        for h in range(24):
            accum[h] = sum(d["activity-by-hour"][h] for d in data.values())

        print_time_table(
            ["received"],
            [accum]
        )

    # Print login report

    if collector["logins"]:
        msg = "User logins per hour"
        print_header(msg)

        data = OrderedDict(sorted(collector["logins"].items(), key=email_sort))

        # Get a list of all of the protocols seen in the logs in reverse count order.
        all_protocols = defaultdict(int)
        for u in data.values():
            for protocol_name, count in u["totals_by_protocol"].items():
                all_protocols[protocol_name] += count
        all_protocols = [k for k, v in sorted(all_protocols.items(), key=lambda kv : -kv[1])]

        print_user_table(
            data.keys(),
            data=[
                (protocol_name, [
                    round(u["totals_by_protocol"][protocol_name] / (u["latest"]-u["earliest"]).total_seconds() * 60*60, 1)
                    if (u["latest"]-u["earliest"]).total_seconds() > 0
                    else 0 # prevent division by zero
                  for u in data.values()])
                for protocol_name in all_protocols
            ],
            sub_data=[
                ("Protocol and Source", [[
                    f"{protocol_name} {host}: {count} times"
                    for (protocol_name, host), count
                    in sorted(u["totals_by_protocol_and_host"].items(), key=lambda kv:-kv[1])
                  ] for u in data.values()])
            ],
            activity=[
                (protocol_name, [u["activity-by-hour"][protocol_name] for u in data.values()])
                for protocol_name in all_protocols
            ],
            earliest=[u["earliest"] for u in data.values()],
            latest=[u["latest"] for u in data.values()],
            numstr=lambda n : str(round(n, 1)),
        )

        accum = { protocol_name: defaultdict(int) for protocol_name in all_protocols }
        for h in range(24):
            for protocol_name in all_protocols:
              accum[protocol_name][h] = sum(d["activity-by-hour"][protocol_name][h] for d in data.values())

        print_time_table(
            all_protocols,
            [accum[protocol_name] for protocol_name in all_protocols]
        )

    if collector["postgrey"]:
        msg = "Greylisted Email {:%Y-%m-%d %H:%M:%S} and {:%Y-%m-%d %H:%M:%S}"
        print_header(msg.format(START_DATE, END_DATE))

        print(textwrap.fill(
            "The following mail was greylisted, meaning the emails were temporarily rejected. "
            "Legitimate senders must try again after three minutes.",
            width=80, initial_indent=" ", subsequent_indent=" "
        ), end='\n\n')

        data = OrderedDict(sorted(collector["postgrey"].items(), key=email_sort))
        users = []
        received = []
        senders = []
        sender_clients = []
        delivered_dates = []

        for recipient in data:
            sorted_recipients = sorted(data[recipient].items(), key=lambda kv: kv[1][0] or kv[1][1])
            for (client_address, sender), (first_date, delivered_date) in sorted_recipients:
                if first_date:
                    users.append(recipient)
                    received.append(first_date)
                    senders.append(sender)
                    delivered_dates.append(delivered_date)
                    sender_clients.append(client_address)

        print_user_table(
            users,
            data=[
                ("received", received),
                ("sender", senders),
                ("delivered", [str(d) or "no retry yet" for d in delivered_dates]),
                ("sending host", sender_clients)
            ],
            delimit=True,
        )

    if collector["rejected"]:
        msg = "Blocked Email {:%Y-%m-%d %H:%M:%S} and {:%Y-%m-%d %H:%M:%S}"
        print_header(msg.format(START_DATE, END_DATE))

        data = OrderedDict(sorted(collector["rejected"].items(), key=email_sort))

        rejects = []

        if VERBOSE:
            for user_data in data.values():
                user_rejects = []
                for date, sender, message in user_data["blocked"]:
                    if len(sender) > 64:
                        sender = sender[:32] + "…" + sender[-32:]
                    user_rejects.extend((f'{date} - {sender} ', '  %s' % message))
                rejects.append(user_rejects)

        print_user_table(
            data.keys(),
            data=[
                ("blocked", [len(u["blocked"]) for u in data.values()]),
            ],
            sub_data=[
                ("blocked emails", rejects),
            ],
            earliest=[u["earliest"] for u in data.values()],
            latest=[u["latest"] for u in data.values()],
        )

    if collector["other-services"] and VERBOSE and False:
        print_header("Other services")
        print("The following unknown services were found in the log file.")
        print(" ", *sorted(collector["other-services"]), sep='\n│ ')


def scan_mail_log_line(line, collector):
    """ Scan a log line and extract interesting data """

    m = re.match(r"(\w+[\s]+\d+ \d+:\d+:\d+) ([\w]+ )?([\w\-/]+)[^:]*: (.*)", line)

    if not m:
        return True

    date, _system, service, log = m.groups()
    collector["scan_count"] += 1

    # print()
    # print("date:", date)
    # print("host:", system)
    # print("service:", service)
    # print("log:", log)

    # Replaced the dateutil parser for a less clever way of parser that is roughly 4 times faster.
    # date = dateutil.parser.parse(date)

    # strptime fails on Feb 29 with ValueError: day is out of range for month if correct year is not provided.
    # See https://bugs.python.org/issue26460
    date = datetime.datetime.strptime(str(NOW.year) + ' ' + date, '%Y %b %d %H:%M:%S')
    # if log date in future, step back a year
    if date > NOW:
      date = date.replace(year = NOW.year - 1)
    #print("date:", date)

    # Check if the found date is within the time span we are scanning
    if date > END_DATE:
        # Don't process, and halt
        return False
    elif date < START_DATE:
        # Don't process, but continue
        return True

    if service == "postfix/submission/smtpd":
        if SCAN_OUT:
            scan_postfix_submission_line(date, log, collector)
    elif service == "postfix/lmtp":
        if SCAN_IN:
            scan_postfix_lmtp_line(date, log, collector)
    elif service.endswith("-login"):
        if SCAN_DOVECOT_LOGIN:
            scan_dovecot_login_line(date, log, collector, service[:4])
    elif service == "postgrey":
        if SCAN_GREY:
            scan_postgrey_line(date, log, collector)
    elif service == "postfix/smtpd":
        if SCAN_BLOCKED:
            scan_postfix_smtpd_line(date, log, collector)
    elif service in {"postfix/qmgr", "postfix/pickup", "postfix/cleanup", "postfix/scache",
                     "spampd", "postfix/anvil", "postfix/master", "opendkim", "postfix/lmtp",
                     "postfix/tlsmgr", "anvil"}:
        # nothing to look at
        return True
    else:
        collector["other-services"].add(service)
        return True

    collector["parse_count"] += 1
    return True


def scan_postgrey_line(date, log, collector):
    """ Scan a postgrey log line and extract interesting data """

    m = re.match(r"action=(greylist|pass), reason=(.*?), (?:delay=\d+, )?client_name=(.*), "
                 "client_address=(.*), sender=(.*), recipient=(.*)",
                 log)

    if m:

        action, reason, client_name, client_address, sender, user = m.groups()

        if user_match(user):

            # Might be useful to group services that use a lot of mail different servers on sub
            # domains like <sub>1.domein.com

            # if '.' in client_name:
            #     addr = client_name.split('.')
            #     if len(addr) > 2:
            #         client_name = '.'.join(addr[1:])

            key = (client_address if client_name == 'unknown' else client_name, sender)

            rep = collector["postgrey"].setdefault(user, {})

            if action == "greylist" and reason == "new":
                rep[key] = (date, rep[key][1] if key in rep else None)
            elif action == "pass":
                rep[key] = (rep[key][0] if key in rep else None, date)


def scan_postfix_smtpd_line(date, log, collector):
    """ Scan a postfix smtpd log line and extract interesting data """

    # Check if the incoming mail was rejected

    m = re.match("NOQUEUE: reject: RCPT from .*?: (.*?); from=<(.*?)> to=<(.*?)>", log)

    if m:
        message, sender, user = m.groups()

        # skip this, if reported in the greylisting report
        if "Recipient address rejected: Greylisted" in message:
            return

        # only log mail to known recipients
        if user_match(user) and (collector["known_addresses"] is None or user in collector["known_addresses"]):
            data = collector["rejected"].get(
                user,
                {
                    "blocked": [],
                    "earliest": None,
                    "latest": None,
                }
            )
            # simplify this one
            m = re.search(
                r"Client host \[(.*?)\] blocked using zen.spamhaus.org; (.*)", message
            )
            if m:
                message = "ip blocked: " + m.group(2)
            else:
                # simplify this one too
                m = re.search(
                    r"Sender address \[.*@(.*)\] blocked using dbl.spamhaus.org; (.*)", message
                )
                if m:
                    message = "domain blocked: " + m.group(2)

            if data["earliest"] is None:
                data["earliest"] = date
            data["latest"] = date
            data["blocked"].append((date, sender, message))

            collector["rejected"][user] = data


def scan_dovecot_login_line(date, log, collector, protocol_name):
    """ Scan a dovecot login log line and extract interesting data """

    m = re.match("Info: Login: user=<(.*?)>, method=PLAIN, rip=(.*?),", log)

    if m:
        # TODO: CHECK DIT
        user, host = m.groups()

        if user_match(user):
            add_login(user, date, protocol_name, host, collector)


def add_login(user, date, protocol_name, host, collector):
            # Get the user data, or create it if the user is new
            data = collector["logins"].get(
                user,
                {
                    "earliest": None,
                    "latest": None,
                    "totals_by_protocol": defaultdict(int),
                    "totals_by_protocol_and_host": defaultdict(int),
                    "activity-by-hour": defaultdict(lambda : defaultdict(int)),
                }
            )

            if data["earliest"] is None:
                data["earliest"] = date
            data["latest"] = date

            data["totals_by_protocol"][protocol_name] += 1
            data["totals_by_protocol_and_host"][(protocol_name, host)] += 1

            if host not in {"127.0.0.1", "::1"} or True:
                data["activity-by-hour"][protocol_name][date.hour] += 1

            collector["logins"][user] = data


def scan_postfix_lmtp_line(date, log, collector):
    """ Scan a postfix lmtp log line and extract interesting data

    It is assumed that every log of postfix/lmtp indicates an email that was successfully
    received by Postfix.

    """

    m = re.match(r"([A-Z0-9]+): to=<(\S+)>, .* Saved", log)

    if m:
        _, user = m.groups()

        if user_match(user):
            # Get the user data, or create it if the user is new
            data = collector["received_mail"].get(
                user,
                {
                    "received_count": 0,
                    "earliest": None,
                    "latest": None,
                    "activity-by-hour": defaultdict(int),
                }
            )

            data["received_count"] += 1
            data["activity-by-hour"][date.hour] += 1

            if data["earliest"] is None:
                data["earliest"] = date
            data["latest"] = date

            collector["received_mail"][user] = data


def scan_postfix_submission_line(date, log, collector):
    """ Scan a postfix submission log line and extract interesting data

    Lines containing a sasl_method with the values PLAIN or LOGIN are assumed to indicate a sent
    email.

    """

    # Match both the 'plain' and 'login' sasl methods, since both authentication methods are
    # allowed by Dovecot. Exclude trailing comma after the username when additional fields
	# follow after.
    m = re.match(r"([A-Z0-9]+): client=(\S+), sasl_method=(PLAIN|LOGIN), sasl_username=(\S+)(?<!,)", log)

    if m:
        _, client, _method, user = m.groups()

        if user_match(user):
            # Get the user data, or create it if the user is new
            data = collector["sent_mail"].get(
                user,
                {
                    "sent_count": 0,
                    "hosts": set(),
                    "earliest": None,
                    "latest": None,
                    "activity-by-hour": defaultdict(int),
                }
            )

            data["sent_count"] += 1
            data["hosts"].add(client)
            data["activity-by-hour"][date.hour] += 1

            if data["earliest"] is None:
                data["earliest"] = date
            data["latest"] = date

            collector["sent_mail"][user] = data

            # Also log this as a login.
            add_login(user, date, "smtp", client, collector)

# Utility functions

def readline(filename):
    """ A generator that returns the lines of a file
    """
    with open(filename, errors='replace', encoding='utf-8') as file:
        while True:
          line = file.readline()
          if not line:
              break
          yield line


def user_match(user):
    """ Check if the given user matches any of the filters """
    return FILTERS is None or any(u in user for u in FILTERS)


def email_sort(email):
    """ Split the given email address into a reverse order tuple, for sorting i.e (domain, name) """
    return tuple(reversed(email[0].split('@')))


def valid_date(string):
    """ Validate the given date string fetched from the --enddate argument """
    try:
        date = dateutil.parser.parse(string)
    except ValueError:
        raise argparse.ArgumentTypeError("Unrecognized date and/or time '%s'" % string)
    return date


# Print functions

def print_time_table(labels, data, do_print=True):
    labels.insert(0, "hour")
    data.insert(0, [str(h) for h in range(24)])

    temp = "│ {:<%d} " % max(len(l) for l in labels)
    lines = [temp.format(label) for label in labels]

    for h in range(24):
        max_len = max(len(str(d[h])) for d in data)
        base = "{:>%d} " % max(2, max_len)

        for i, d in enumerate(data):
            lines[i] += base.format(d[h])

    lines.insert(0, "┬ totals by time of day:")
    lines.append("└" + (len(lines[-1]) - 2) * "─")

    if do_print:
        print("\n".join(lines))
        return None
    else:
        return lines


def print_user_table(users, data=None, sub_data=None, activity=None, latest=None, earliest=None,
                     delimit=False, numstr=str):
    str_temp = "{:<32} "
    lines = []
    data = data or []

    col_widths = len(data) * [0]
    col_left = len(data) * [False]
    vert_pos = 0

    do_accum = all(isinstance(n, (int, float)) for _, d in data for n in d)
    data_accum = len(data) * ([0] if do_accum else [" "])

    last_user = None

    for row, user in enumerate(users):

        if delimit:
            if last_user and last_user != user:
                lines.append(len(lines[-1]) * "…")
            last_user = user

        line = "{:<32} ".format(user[:31] + "…" if len(user) > 32 else user)

        for col, (l, d) in enumerate(data):
            if isinstance(d[row], str):
                col_str = str_temp.format(d[row][:31] + "…" if len(d[row]) > 32 else d[row])
                col_left[col] = True
            elif isinstance(d[row], datetime.datetime):
                col_str = f"{d[row]!s:<20}"
                col_left[col] = True
            else:
                temp = "{:>%s}" % max(5, len(l) + 1, len(str(d[row])) + 1)
                col_str = temp.format(str(d[row]))
            col_widths[col] = max(col_widths[col], len(col_str))
            line += col_str

            if do_accum:
                data_accum[col] += d[row]

        try:
            if None not in [latest, earliest]: # noqa PLR6201
                vert_pos = len(line)
                e = earliest[row]
                l = latest[row]
                timespan = relativedelta(l, e)
                if timespan.months:
                    temp = " │ {:0.1f} months"
                    line += temp.format(timespan.months + timespan.days / 30.0)
                elif timespan.days:
                    temp = " │ {:0.1f} days"
                    line += temp.format(timespan.days + timespan.hours / 24.0)
                elif (e.hour, e.minute) == (l.hour, l.minute):
                    temp = " │ {:%H:%M}"
                    line += temp.format(e)
                else:
                    temp = " │ {:%H:%M} - {:%H:%M}"
                    line += temp.format(e, l)

        except KeyError:
            pass

        lines.append(line.rstrip())

        try:
            if VERBOSE:
                if sub_data is not None:
                    for l, d in sub_data:
                        if d[row]:
                            lines.extend(('┬', '│ %s' % l, '├─%s─' % (len(l) * '─'), '│'))
                            max_len = 0
                            for v in list(d[row]):
                                lines.append("│ %s" % v)
                                max_len = max(max_len, len(v))
                            lines.append("└" + (max_len + 1) * "─")

                if activity is not None:
                    lines.extend(print_time_table(
                        [label for label, _ in activity],
                        [data[row] for _, data in activity],
                        do_print=False
                    ))

        except KeyError:
            pass

    header = str_temp.format("")

    for col, (l, _) in enumerate(data):
        if col_left[col]:
            header += l.ljust(max(5, len(l) + 1, col_widths[col]))
        else:
            header += l.rjust(max(5, len(l) + 1, col_widths[col]))

    if None not in [latest, earliest]: # noqa PLR6201
        header += " │ timespan   "

    lines.insert(0, header.rstrip())

    table_width = max(len(l) for l in lines)
    t_line = table_width * "─"
    b_line = table_width * "─"

    if vert_pos:
        t_line = t_line[:vert_pos + 1] + "┼" + t_line[vert_pos + 2:]
        b_line = b_line[:vert_pos + 1] + ("┬" if VERBOSE else "┼") + b_line[vert_pos + 2:]

    lines.insert(1, t_line)
    lines.append(b_line)

    # Print totals

    data_accum = [numstr(a) for a in data_accum]
    footer = str_temp.format("Totals:" if do_accum else " ")
    for row, (l, _) in enumerate(data):
        temp = "{:>%d}" % max(5, len(l) + 1)
        footer += temp.format(data_accum[row])

    try:
        if None not in [latest, earliest]: # noqa PLR6201
            max_l = max(latest)
            min_e = min(earliest)
            timespan = relativedelta(max_l, min_e)
            if timespan.days:
                temp = " │ {:0.2f} days"
                footer += temp.format(timespan.days + timespan.hours / 24.0)
            elif (min_e.hour, min_e.minute) == (max_l.hour, max_l.minute):
                temp = " │ {:%H:%M}"
                footer += temp.format(min_e)
            else:
                temp = " │ {:%H:%M} - {:%H:%M}"
                footer += temp.format(min_e, max_l)

    except KeyError:
        pass

    lines.append(footer)

    print("\n".join(lines))


def print_header(msg):
    print('\n' + msg)
    print("═" * len(msg), '\n')


if __name__ == "__main__":
    try:
        env_vars = utils.load_environment()
    except FileNotFoundError:
        env_vars = {}

    parser = argparse.ArgumentParser(
        description="Scan the mail log files for interesting data. By default, this script "
                    "shows today's incoming and outgoing mail statistics. This script was ("
                    "re)written for the Mail-in-a-box email server."
                    "https://github.com/mail-in-a-box/mailinabox",
        add_help=False
    )

    # Switches to determine what to parse and what to ignore

    parser.add_argument("-r", "--received", help="Scan for received emails.",
                        action="store_true")
    parser.add_argument("-s", "--sent", help="Scan for sent emails.",
                        action="store_true")
    parser.add_argument("-l", "--logins", help="Scan for user logins to IMAP/POP3.",
                        action="store_true")
    parser.add_argument("-g", "--grey", help="Scan for greylisted emails.",
                        action="store_true")
    parser.add_argument("-b", "--blocked", help="Scan for blocked emails.",
                        action="store_true")

    parser.add_argument("-t", "--timespan", choices=TIME_DELTAS.keys(), default='today',
                        metavar='<time span>',
                        help="Time span to scan, going back from the end date. Possible values: "
                             "{}. Defaults to 'today'.".format(", ".join(list(TIME_DELTAS.keys()))))
    # keep the --startdate arg for backward compatibility
    parser.add_argument("-d", "--enddate", "--startdate",  action="store", dest="enddate",
                        type=valid_date, metavar='<end date>',
                        help="Date and time to end scanning the log file. If no date is "
                             "provided, scanning will end at the current date and time. "
                             "Alias --startdate is for compatibility.")
    parser.add_argument("-u", "--users", action="store", dest="users",
                        metavar='<email1,email2,email...>',
                        help="Comma separated list of (partial) email addresses to filter the "
                             "output with.")

    parser.add_argument('-h', '--help', action='help', help="Print this message and exit.")
    parser.add_argument("-v", "--verbose", help="Output extra data where available.",
                        action="store_true")

    args = parser.parse_args()

    if args.enddate is not None:
        END_DATE = args.enddate
        if args.timespan == 'today':
            args.timespan = 'day'
        print(f"Setting end date to {END_DATE}")

    START_DATE = END_DATE - TIME_DELTAS[args.timespan]

    VERBOSE = args.verbose

    if args.received or args.sent or args.logins or args.grey or args.blocked:
        SCAN_IN = args.received
        if not SCAN_IN:
            print("Ignoring received emails")

        SCAN_OUT = args.sent
        if not SCAN_OUT:
            print("Ignoring sent emails")

        SCAN_DOVECOT_LOGIN = args.logins
        if not SCAN_DOVECOT_LOGIN:
            print("Ignoring logins")

        SCAN_GREY = args.grey
        if SCAN_GREY:
            print("Showing greylisted emails")

        SCAN_BLOCKED = args.blocked
        if SCAN_BLOCKED:
            print("Showing blocked emails")

    if args.users is not None:
        FILTERS = args.users.strip().split(',')

    scan_mail_log(env_vars)



File: management/web_update.py

# Creates an nginx configuration file so we serve HTTP/HTTPS on all
# domains for which a mail account has been set up.
########################################################################

import os.path, re, rtyaml

from mailconfig import get_mail_domains
from dns_update import get_custom_dns_config, get_dns_zones
from ssl_certificates import get_ssl_certificates, get_domain_ssl_files, check_certificate
from utils import shell, safe_domain_name, sort_domains

def get_web_domains(env, include_www_redirects=True, include_auto=True, exclude_dns_elsewhere=True):
	# What domains should we serve HTTP(S) for?
	domains = set()

	# Serve web for all mail domains so that we might at least
	# provide auto-discover of email settings, and also a static website
	# if the user wants to make one.
	domains |= get_mail_domains(env)

	if include_www_redirects and include_auto:
		# Add 'www.' subdomains that we want to provide default redirects
		# to the main domain for. We'll add 'www.' to any DNS zones, i.e.
		# the topmost of each domain we serve.
		domains |= {'www.' + zone for zone, zonefile in get_dns_zones(env)}

	if include_auto:
		# Add Autoconfiguration domains for domains that there are user accounts at:
		# 'autoconfig.' for Mozilla Thunderbird auto setup.
		# 'autodiscover.' for ActiveSync autodiscovery (Z-Push).
		domains |= {'autoconfig.' + maildomain for maildomain in get_mail_domains(env, users_only=True)}
		domains |= {'autodiscover.' + maildomain for maildomain in get_mail_domains(env, users_only=True)}

		# 'mta-sts.' for MTA-STS support for all domains that have email addresses.
		domains |= {'mta-sts.' + maildomain for maildomain in get_mail_domains(env)}

	if exclude_dns_elsewhere:
		# ...Unless the domain has an A/AAAA record that maps it to a different
		# IP address than this box. Remove those domains from our list.
		domains -= get_domains_with_a_records(env)

	# Ensure the PRIMARY_HOSTNAME is in the list so we can serve webmail
	# as well as Z-Push for Exchange ActiveSync. This can't be removed
	# by a custom A/AAAA record and is never a 'www.' redirect.
	domains.add(env['PRIMARY_HOSTNAME'])

	# Sort the list so the nginx conf gets written in a stable order.
	return sort_domains(domains, env)


def get_domains_with_a_records(env):
	domains = set()
	dns = get_custom_dns_config(env)
	for domain, rtype, value in dns:
		if rtype == "CNAME" or (rtype in {"A", "AAAA"} and value not in {"local", env['PUBLIC_IP']}):
			domains.add(domain)
	return domains

def get_web_domains_with_root_overrides(env):
	# Load custom settings so we can tell what domains have a redirect or proxy set up on '/',
	# which means static hosting is not happening.
	root_overrides = { }
	nginx_conf_custom_fn = os.path.join(env["STORAGE_ROOT"], "www/custom.yaml")
	if os.path.exists(nginx_conf_custom_fn):
		with open(nginx_conf_custom_fn, encoding='utf-8') as f:
			custom_settings = rtyaml.load(f)
		for domain, settings in custom_settings.items():
			for type, value in [('redirect', settings.get('redirects', {}).get('/')),
				('proxy', settings.get('proxies', {}).get('/'))]:
				if value:
					root_overrides[domain] = (type, value)
	return root_overrides

def do_web_update(env):
	# Pre-load what SSL certificates we will use for each domain.
	ssl_certificates = get_ssl_certificates(env)

	# Helper for reading config files and templates
	def read_conf(conf_fn):
		with open(os.path.join(os.path.dirname(__file__), "../conf", conf_fn), encoding='utf-8') as f:
			return f.read()

	# Build an nginx configuration file.
	nginx_conf = read_conf("nginx-top.conf")

	# Load the templates.
	template0 = read_conf("nginx.conf")
	template1 = read_conf("nginx-alldomains.conf")
	template2 = read_conf("nginx-primaryonly.conf")
	template3 = "\trewrite ^(.*) https://$REDIRECT_DOMAIN$1 permanent;\n"

	# Add the PRIMARY_HOST configuration first so it becomes nginx's default server.
	nginx_conf += make_domain_config(env['PRIMARY_HOSTNAME'], [template0, template1, template2], ssl_certificates, env)

	# Add configuration all other web domains.
	has_root_proxy_or_redirect = get_web_domains_with_root_overrides(env)
	web_domains_not_redirect = get_web_domains(env, include_www_redirects=False)
	for domain in get_web_domains(env):
		if domain == env['PRIMARY_HOSTNAME']:
			# PRIMARY_HOSTNAME is handled above.
			continue
		if domain in web_domains_not_redirect:
			# This is a regular domain.
			if domain not in has_root_proxy_or_redirect:
				nginx_conf += make_domain_config(domain, [template0, template1], ssl_certificates, env)
			else:
				nginx_conf += make_domain_config(domain, [template0], ssl_certificates, env)
		else:
			# Add default 'www.' redirect.
			nginx_conf += make_domain_config(domain, [template0, template3], ssl_certificates, env)

	# Did the file change? If not, don't bother writing & restarting nginx.
	nginx_conf_fn = "/etc/nginx/conf.d/local.conf"
	if os.path.exists(nginx_conf_fn):
		with open(nginx_conf_fn, encoding='utf-8') as f:
			if f.read() == nginx_conf:
				return ""

	# Save the file.
	with open(nginx_conf_fn, "w", encoding='utf-8') as f:
		f.write(nginx_conf)

	# Kick nginx. Since this might be called from the web admin
	# don't do a 'restart'. That would kill the connection before
	# the API returns its response. A 'reload' should be good
	# enough and doesn't break any open connections.
	shell('check_call', ["/usr/sbin/service", "nginx", "reload"])

	return "web updated\n"

def make_domain_config(domain, templates, ssl_certificates, env):
	# GET SOME VARIABLES

	# Where will its root directory be for static files?
	root = get_web_root(domain, env)

	# What private key and SSL certificate will we use for this domain?
	tls_cert = get_domain_ssl_files(domain, ssl_certificates, env)

	# ADDITIONAL DIRECTIVES.

	nginx_conf_extra = ""

	# Because the certificate may change, we should recognize this so we
	# can trigger an nginx update.
	def hashfile(filepath):
		import hashlib
		sha1 = hashlib.sha1()
		with open(filepath, 'rb') as f:
			sha1.update(f.read())
		return sha1.hexdigest()
	nginx_conf_extra += "\t# ssl files sha1: {} / {}\n".format(hashfile(tls_cert["private-key"]), hashfile(tls_cert["certificate"]))

	# Add in any user customizations in YAML format.
	hsts = "yes"
	nginx_conf_custom_fn = os.path.join(env["STORAGE_ROOT"], "www/custom.yaml")
	if os.path.exists(nginx_conf_custom_fn):
		with open(nginx_conf_custom_fn, encoding='utf-8') as f:
			yaml = rtyaml.load(f)
		if domain in yaml:
			yaml = yaml[domain]

			# any proxy or redirect here?
			for path, url in yaml.get("proxies", {}).items():
				# Parse some flags in the fragment of the URL.
				pass_http_host_header = False
				proxy_redirect_off = False
				frame_options_header_sameorigin = False
				web_sockets = False
				m = re.search("#(.*)$", url)
				if m:
					for flag in m.group(1).split(","):
						if flag == "pass-http-host":
							pass_http_host_header = True
						elif flag == "no-proxy-redirect":
							proxy_redirect_off = True
						elif flag == "frame-options-sameorigin":
							frame_options_header_sameorigin = True
						elif flag == "web-sockets":
							web_sockets = True
					url = re.sub("#(.*)$", "", url)

				nginx_conf_extra += "\tlocation %s {" % path
				nginx_conf_extra += "\n\t\tproxy_pass %s;" % url
				if proxy_redirect_off:
					nginx_conf_extra += "\n\t\tproxy_redirect off;"
				if pass_http_host_header:
					nginx_conf_extra += "\n\t\tproxy_set_header Host $http_host;"
				if frame_options_header_sameorigin:
					nginx_conf_extra += "\n\t\tproxy_set_header X-Frame-Options SAMEORIGIN;"
				if web_sockets:
					nginx_conf_extra += "\n\t\tproxy_http_version 1.1;"
					nginx_conf_extra += "\n\t\tproxy_set_header Upgrade $http_upgrade;"
					nginx_conf_extra += "\n\t\tproxy_set_header Connection 'Upgrade';"
				nginx_conf_extra += "\n\t\tproxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;"
				nginx_conf_extra += "\n\t\tproxy_set_header X-Forwarded-Host $http_host;"
				nginx_conf_extra += "\n\t\tproxy_set_header X-Forwarded-Proto $scheme;"
				nginx_conf_extra += "\n\t\tproxy_set_header X-Real-IP $remote_addr;"
				nginx_conf_extra += "\n\t}\n"
			for path, alias in yaml.get("aliases", {}).items():
				nginx_conf_extra += "\tlocation %s {" % path
				nginx_conf_extra += "\n\t\talias %s;" % alias
				nginx_conf_extra += "\n\t}\n"
			for path, url in yaml.get("redirects", {}).items():
				nginx_conf_extra += f"\trewrite {path} {url} permanent;\n"

			# override the HSTS directive type
			hsts = yaml.get("hsts", hsts)

	# Add the HSTS header.
	if hsts == "yes":
		nginx_conf_extra += '\tadd_header Strict-Transport-Security "max-age=15768000" always;\n'
	elif hsts == "preload":
		nginx_conf_extra += '\tadd_header Strict-Transport-Security "max-age=15768000; includeSubDomains; preload" always;\n'

	# Add in any user customizations in the includes/ folder.
	nginx_conf_custom_include = os.path.join(env["STORAGE_ROOT"], "www", safe_domain_name(domain) + ".conf")
	if os.path.exists(nginx_conf_custom_include):
		nginx_conf_extra += "\tinclude %s;\n" % (nginx_conf_custom_include)
	# PUT IT ALL TOGETHER

	# Combine the pieces. Iteratively place each template into the "# ADDITIONAL DIRECTIVES HERE" placeholder
	# of the previous template.
	nginx_conf = "# ADDITIONAL DIRECTIVES HERE\n"
	for t in [*templates, nginx_conf_extra]:
		nginx_conf = re.sub("[ \t]*# ADDITIONAL DIRECTIVES HERE *\n", t, nginx_conf)

	# Replace substitution strings in the template & return.
	nginx_conf = nginx_conf.replace("$STORAGE_ROOT", env['STORAGE_ROOT'])
	nginx_conf = nginx_conf.replace("$HOSTNAME", domain)
	nginx_conf = nginx_conf.replace("$ROOT", root)
	nginx_conf = nginx_conf.replace("$SSL_KEY", tls_cert["private-key"])
	nginx_conf = nginx_conf.replace("$SSL_CERTIFICATE", tls_cert["certificate"])
	return nginx_conf.replace("$REDIRECT_DOMAIN", re.sub(r"^www\.", "", domain)) # for default www redirects to parent domain


def get_web_root(domain, env, test_exists=True):
	# Try STORAGE_ROOT/web/domain_name if it exists, but fall back to STORAGE_ROOT/web/default.
	for test_domain in (domain, 'default'):
		root = os.path.join(env["STORAGE_ROOT"], "www", safe_domain_name(test_domain))
		if os.path.exists(root) or not test_exists: break
	return root

def get_web_domains_info(env):
	www_redirects = set(get_web_domains(env)) - set(get_web_domains(env, include_www_redirects=False))
	has_root_proxy_or_redirect = set(get_web_domains_with_root_overrides(env))
	ssl_certificates = get_ssl_certificates(env)

	# for the SSL config panel, get cert status
	def check_cert(domain):
		try:
			tls_cert = get_domain_ssl_files(domain, ssl_certificates, env, allow_missing_cert=True)
		except OSError: # PRIMARY_HOSTNAME cert is missing
			tls_cert = None
		if tls_cert is None: return ("danger", "No certificate installed.")
		cert_status, cert_status_details = check_certificate(domain, tls_cert["certificate"], tls_cert["private-key"])
		if cert_status == "OK":
			return ("success", "Signed & valid. " + cert_status_details)
		elif cert_status == "SELF-SIGNED":
			return ("warning", "Self-signed. Get a signed certificate to stop warnings.")
		else:
			return ("danger", "Certificate has a problem: " + cert_status)

	return [
		{
			"domain": domain,
			"root": get_web_root(domain, env),
			"custom_root": get_web_root(domain, env, test_exists=False),
			"ssl_certificate": check_cert(domain),
			"static_enabled": domain not in (www_redirects | has_root_proxy_or_redirect),
		}
		for domain in get_web_domains(env)
	]




File: management/cli.py

#!/usr/bin/python3
#
# This is a command-line script for calling management APIs
# on the Mail-in-a-Box control panel backend. The script
# reads /var/lib/mailinabox/api.key for the backend's
# root API key. This file is readable only by root, so this
# tool can only be used as root.

import sys, getpass, urllib.request, urllib.error, json, csv
import contextlib

def mgmt(cmd, data=None, is_json=False):
	# The base URL for the management daemon. (Listens on IPv4 only.)
	mgmt_uri = 'http://127.0.0.1:10222'

	setup_key_auth(mgmt_uri)

	req = urllib.request.Request(mgmt_uri + cmd, urllib.parse.urlencode(data).encode("utf8") if data else None)
	try:
		response = urllib.request.urlopen(req)
	except urllib.error.HTTPError as e:
		if e.code == 401:
			with contextlib.suppress(Exception):
				print(e.read().decode("utf8"))
			print("The management daemon refused access. The API key file may be out of sync. Try 'service mailinabox restart'.", file=sys.stderr)
		elif hasattr(e, 'read'):
			print(e.read().decode('utf8'), file=sys.stderr)
		else:
			print(e, file=sys.stderr)
		sys.exit(1)
	resp = response.read().decode('utf8')
	if is_json: resp = json.loads(resp)
	return resp

def read_password():
    while True:
        first = getpass.getpass('password: ')
        if len(first) < 8:
            print("Passwords must be at least eight characters.")
            continue
        second = getpass.getpass(' (again): ')
        if first != second:
            print("Passwords not the same. Try again.")
            continue
        break
    return first

def setup_key_auth(mgmt_uri):
	with open('/var/lib/mailinabox/api.key', encoding='utf-8') as f:
		key = f.read().strip()

	auth_handler = urllib.request.HTTPBasicAuthHandler()
	auth_handler.add_password(
		realm='Mail-in-a-Box Management Server',
		uri=mgmt_uri,
		user=key,
		passwd='')
	opener = urllib.request.build_opener(auth_handler)
	urllib.request.install_opener(opener)

if len(sys.argv) < 2:
	print("""Usage:
  {cli} user                                     (lists users)
  {cli} user add user@domain.com [password]
  {cli} user password user@domain.com [password]
  {cli} user remove user@domain.com
  {cli} user make-admin user@domain.com
  {cli} user remove-admin user@domain.com
  {cli} user admins                              (lists admins)
  {cli} user mfa show user@domain.com            (shows MFA devices for user, if any)
  {cli} user mfa disable user@domain.com [id]    (disables MFA for user)
  {cli} alias                                    (lists aliases)
  {cli} alias add incoming.name@domain.com sent.to@other.domain.com
  {cli} alias add incoming.name@domain.com 'sent.to@other.domain.com, multiple.people@other.domain.com'
  {cli} alias remove incoming.name@domain.com

Removing a mail user does not delete their mail folders on disk. It only prevents IMAP/SMTP login.
""".format(
	cli="management/cli.py"
		))

elif sys.argv[1] == "user" and len(sys.argv) == 2:
	# Dump a list of users, one per line. Mark admins with an asterisk.
	users = mgmt("/mail/users?format=json", is_json=True)
	for domain in users:
		for user in domain["users"]:
			if user['status'] == 'inactive': continue
			print(user['email'], end='')
			if "admin" in user['privileges']:
				print("*", end='')
			print()

elif sys.argv[1] == "user" and sys.argv[2] in {"add", "password"}:
	if len(sys.argv) < 5:
		email = input('email: ') if len(sys.argv) < 4 else sys.argv[3]
		pw = read_password()
	else:
		email, pw = sys.argv[3:5]

	if sys.argv[2] == "add":
		print(mgmt("/mail/users/add", { "email": email, "password": pw }))
	elif sys.argv[2] == "password":
		print(mgmt("/mail/users/password", { "email": email, "password": pw }))

elif sys.argv[1] == "user" and sys.argv[2] == "remove" and len(sys.argv) == 4:
	print(mgmt("/mail/users/remove", { "email": sys.argv[3] }))

elif sys.argv[1] == "user" and sys.argv[2] in {"make-admin", "remove-admin"} and len(sys.argv) == 4:
	action = 'add' if sys.argv[2] == 'make-admin' else 'remove'
	print(mgmt("/mail/users/privileges/" + action, { "email": sys.argv[3], "privilege": "admin" }))

elif sys.argv[1] == "user" and sys.argv[2] == "admins":
	# Dump a list of admin users.
	users = mgmt("/mail/users?format=json", is_json=True)
	for domain in users:
		for user in domain["users"]:
			if "admin" in user['privileges']:
				print(user['email'])

elif sys.argv[1] == "user" and len(sys.argv) == 5 and sys.argv[2:4] == ["mfa", "show"]:
	# Show MFA status for a user.
	status = mgmt("/mfa/status", { "user": sys.argv[4] }, is_json=True)
	W = csv.writer(sys.stdout)
	W.writerow(["id", "type", "label"])
	for mfa in status["enabled_mfa"]:
		W.writerow([mfa["id"], mfa["type"], mfa["label"]])

elif sys.argv[1] == "user" and len(sys.argv) in {5, 6} and sys.argv[2:4] == ["mfa", "disable"]:
	# Disable MFA (all or a particular device) for a user.
	print(mgmt("/mfa/disable", { "user": sys.argv[4], "mfa-id": sys.argv[5] if len(sys.argv) == 6 else None }))

elif sys.argv[1] == "alias" and len(sys.argv) == 2:
	print(mgmt("/mail/aliases"))

elif sys.argv[1] == "alias" and sys.argv[2] == "add" and len(sys.argv) == 5:
	print(mgmt("/mail/aliases/add", { "address": sys.argv[3], "forwards_to": sys.argv[4] }))

elif sys.argv[1] == "alias" and sys.argv[2] == "remove" and len(sys.argv) == 4:
	print(mgmt("/mail/aliases/remove", { "address": sys.argv[3] }))

else:
	print("Invalid command-line arguments.")
	sys.exit(1)




File: management/utils.py

import os.path

# DO NOT import non-standard modules. This module is imported by
# migrate.py which runs on fresh machines before anything is installed
# besides Python.

# THE ENVIRONMENT FILE AT /etc/mailinabox.conf

def load_environment():
    # Load settings from /etc/mailinabox.conf.
    return load_env_vars_from_file("/etc/mailinabox.conf")

def load_env_vars_from_file(fn):
    # Load settings from a KEY=VALUE file.
    import collections
    env = collections.OrderedDict()
    with open(fn, encoding="utf-8")  as f:
        for line in f:
            env.setdefault(*line.strip().split("=", 1))
    return env

def save_environment(env):
    with open("/etc/mailinabox.conf", "w", encoding="utf-8") as f:
        for k, v in env.items():
            f.write(f"{k}={v}\n")

# THE SETTINGS FILE AT STORAGE_ROOT/settings.yaml.

def write_settings(config, env):
    import rtyaml
    fn = os.path.join(env['STORAGE_ROOT'], 'settings.yaml')
    with open(fn, "w", encoding="utf-8") as f:
        f.write(rtyaml.dump(config))

def load_settings(env):
    import rtyaml
    fn = os.path.join(env['STORAGE_ROOT'], 'settings.yaml')
    try:
        with open(fn, encoding="utf-8") as f:
            config = rtyaml.load(f)
        if not isinstance(config, dict): raise ValueError # caught below
        return config
    except:
        return { }

# UTILITIES

def safe_domain_name(name):
    # Sanitize a domain name so it is safe to use as a file name on disk.
    import urllib.parse
    return urllib.parse.quote(name, safe='')

def sort_domains(domain_names, env):
    # Put domain names in a nice sorted order.

    # The nice order will group domain names by DNS zone, i.e. the top-most
    # domain name that we serve that ecompasses a set of subdomains. Map
    # each of the domain names to the zone that contains them. Walk the domains
    # from shortest to longest since zones are always shorter than their
    # subdomains.
    zones = { }
    for domain in sorted(domain_names, key=len):
        for z in zones.values():
            if domain.endswith("." + z):
                # We found a parent domain already in the list.
                zones[domain] = z
                break
        else:
            # 'break' did not occur: there is no parent domain, so it is its
            # own zone.
            zones[domain] = domain

    # Sort the zones.
    zone_domains = sorted(zones.values(),
      key = lambda d : (
        # PRIMARY_HOSTNAME or the zone that contains it is always first.
        not (d == env['PRIMARY_HOSTNAME'] or env['PRIMARY_HOSTNAME'].endswith("." + d)),

        # Then just dumb lexicographically.
        d,
      ))

    # Now sort the domain names that fall within each zone.
    return sorted(domain_names,
      key = lambda d : (
        # First by zone.
        zone_domains.index(zones[d]),

        # PRIMARY_HOSTNAME is always first within the zone that contains it.
        d != env['PRIMARY_HOSTNAME'],

        # Followed by any of its subdomains.
        not d.endswith("." + env['PRIMARY_HOSTNAME']),

        # Then in right-to-left lexicographic order of the .-separated parts of the name.
        list(reversed(d.split("."))),
      ))


def sort_email_addresses(email_addresses, env):
    email_addresses = set(email_addresses)
    domains = {email.split("@", 1)[1] for email in email_addresses if "@" in email}
    ret = []
    for domain in sort_domains(domains, env):
        domain_emails = {email for email in email_addresses if email.endswith("@" + domain)}
        ret.extend(sorted(domain_emails))
        email_addresses -= domain_emails
    ret.extend(sorted(email_addresses)) # whatever is left
    return ret

def shell(method, cmd_args, env=None, capture_stderr=False, return_bytes=False, trap=False, input=None):
    # A safe way to execute processes.
    # Some processes like apt-get require being given a sane PATH.
    import subprocess

    if env is None:
        env = {}
    env.update({ "PATH": "/sbin:/bin:/usr/sbin:/usr/bin" })
    kwargs = {
        'env': env,
        'stderr': None if not capture_stderr else subprocess.STDOUT,
    }
    if method == "check_output" and input is not None:
        kwargs['input'] = input

    if not trap:
        ret = getattr(subprocess, method)(cmd_args, **kwargs)
    else:
        try:
            ret = getattr(subprocess, method)(cmd_args, **kwargs)
            code = 0
        except subprocess.CalledProcessError as e:
            ret = e.output
            code = e.returncode
    if not return_bytes and isinstance(ret, bytes): ret = ret.decode("utf8")
    if not trap:
        return ret
    else:
        return code, ret

def create_syslog_handler():
    import logging.handlers
    handler = logging.handlers.SysLogHandler(address='/dev/log')
    handler.setLevel(logging.WARNING)
    return handler

def du(path):
    # Computes the size of all files in the path, like the `du` command.
    # Based on http://stackoverflow.com/a/17936789. Takes into account
    # soft and hard links.
    total_size = 0
    seen = set()
    for dirpath, _dirnames, filenames in os.walk(path):
        for f in filenames:
            fp = os.path.join(dirpath, f)
            try:
                stat = os.lstat(fp)
            except OSError:
                continue
            if stat.st_ino in seen:
                continue
            seen.add(stat.st_ino)
            total_size += stat.st_size
    return total_size

def wait_for_service(port, public, env, timeout):
	# Block until a service on a given port (bound privately or publicly)
	# is taking connections, with a maximum timeout.
	import socket, time
	start = time.perf_counter()
	while True:
		s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
		s.settimeout(timeout/3)
		try:
			s.connect(("127.0.0.1" if not public else env['PUBLIC_IP'], port))
			return True
		except OSError:
			if time.perf_counter() > start+timeout:
				return False
		time.sleep(min(timeout/4, 1))

def get_ssh_port():
	port_value = get_ssh_config_value("port")

	if port_value:
		return int(port_value)

	return None

def get_ssh_config_value(parameter_name):
	# Returns ssh configuration value for the provided parameter
	try:
		output = shell('check_output', ['sshd', '-T'])
	except FileNotFoundError:
		# sshd is not installed. That's ok.
		return None
	except subprocess.CalledProcessError:
		# error while calling shell command
		return None

	for line in output.split("\n"):
		if " " not in line: continue # there's a blank line at the end
		key, values = line.split(" ", 1)
		if key == parameter_name:
			return values # space-delimited if there are multiple values

	# Did not find the parameter!
	return None

if __name__ == "__main__":
	from web_update import get_web_domains
	env = load_environment()
	domains = get_web_domains(env)
	for domain in domains:
		print(domain)



File: management/email_administrator.py

#!/usr/local/lib/mailinabox/env/bin/python

# Reads in STDIN. If the stream is not empty, mail it to the system administrator.

import sys

import html
import smtplib

from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText

# In Python 3.6:
#from email.message import Message

from utils import load_environment

# Load system environment info.
env = load_environment()

# Process command line args.
subject = sys.argv[1]

# Administrator's email address.
admin_addr = "administrator@" + env['PRIMARY_HOSTNAME']

# Read in STDIN.
content = sys.stdin.read().strip()

# If there's nothing coming in, just exit.
if content == "":
    sys.exit(0)

# create MIME message
msg = MIMEMultipart('alternative')

# In Python 3.6:
#msg = Message()

msg['From'] = '"{}" <{}>'.format(env['PRIMARY_HOSTNAME'], admin_addr)
msg['To'] = admin_addr
msg['Subject'] = "[{}] {}".format(env['PRIMARY_HOSTNAME'], subject)

content_html = f'<html><body><pre style="overflow-x: scroll; white-space: pre;">{html.escape(content)}</pre></body></html>'

msg.attach(MIMEText(content, 'plain'))
msg.attach(MIMEText(content_html, 'html'))

# In Python 3.6:
#msg.set_content(content)
#msg.add_alternative(content_html, "html")

# send
smtpclient = smtplib.SMTP('127.0.0.1', 25)
smtpclient.ehlo()
smtpclient.sendmail(
        admin_addr, # MAIL FROM
        admin_addr, # RCPT TO
        msg.as_string())
smtpclient.quit()



File: management/mailconfig.py

#!/usr/local/lib/mailinabox/env/bin/python

# NOTE:
# This script is run both using the system-wide Python 3
# interpreter (/usr/bin/python3) as well as through the
# virtualenv (/usr/local/lib/mailinabox/env). So only
# import packages at the top level of this script that
# are installed in *both* contexts. We use the system-wide
# Python 3 in setup/questions.sh to validate the email
# address entered by the user.

import os, sqlite3, re
import utils
from email_validator import validate_email as validate_email_, EmailNotValidError
import idna

def validate_email(email, mode=None):
	# Checks that an email address is syntactically valid. Returns True/False.
	# An email address may contain ASCII characters only because Dovecot's
	# authentication mechanism gets confused with other character encodings.
	#
	# When mode=="user", we're checking that this can be a user account name.
	# Dovecot has tighter restrictions - letters, numbers, underscore, and
	# dash only!
	#
	# When mode=="alias", we're allowing anything that can be in a Postfix
	# alias table, i.e. omitting the local part ("@domain.tld") is OK.

	# Check the syntax of the address.
	try:
		validate_email_(email,
			allow_smtputf8=False,
			check_deliverability=False,
			allow_empty_local=(mode=="alias")
			)
	except EmailNotValidError:
		return False

	if mode == 'user':
		# There are a lot of characters permitted in email addresses, but
		# Dovecot's sqlite auth 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.
		# Our database is case sensitive (oops), which affects mail delivery
		# (Postfix always queries in lowercase?), so also only permit lowercase
		# letters.
		if len(email) > 255: return False
		if re.search(r'[^\@\.a-z0-9_\-]+', email):
			return False

	# Everything looks good.
	return True

def sanitize_idn_email_address(email):
	# The user may enter Unicode in an email address. Convert the domain part
	# to IDNA before going into our database. Leave the local part alone ---
	# although validate_email will reject non-ASCII characters.
	#
	# The domain name system only exists in ASCII, so it doesn't make sense
	# to store domain names in Unicode. We want to store what is meaningful
	# to the underlying protocols.
	try:
		localpart, domainpart = email.split("@")
		domainpart = idna.encode(domainpart).decode('ascii')
		return localpart + "@" + domainpart
	except (ValueError, idna.IDNAError):
		# ValueError: String does not have a single @-sign, so it is not
		# a valid email address. IDNAError: Domain part is not IDNA-valid.
		# Validation is not this function's job, so return value unchanged.
		# If there are non-ASCII characters it will be filtered out by
		# validate_email.
		return email

def prettify_idn_email_address(email):
	# This is the opposite of sanitize_idn_email_address. We store domain
	# names in IDNA in the database, but we want to show Unicode to the user.
	try:
		localpart, domainpart = email.split("@")
		domainpart = idna.decode(domainpart.encode("ascii"))
		return localpart + "@" + domainpart
	except (ValueError, UnicodeError, idna.IDNAError):
		# Failed to decode IDNA, or the email address does not have a
		# single @-sign. Should never happen.
		return email

def is_dcv_address(email):
	email = email.lower()
	return any(email.startswith((localpart + "@", localpart + "+")) for localpart in ("admin", "administrator", "postmaster", "hostmaster", "webmaster", "abuse"))

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):
	# Returns a flat, sorted list of all user accounts.
	c = open_database(env)
	c.execute('SELECT email FROM users')
	users = [ row[0] for row in c.fetchall() ]
	return utils.sort_email_addresses(users, env)

def get_mail_users_ex(env, with_archived=False):
	# Returns a complex data structure of all user accounts, optionally
	# including archived (status="inactive") accounts.
	#
	# [
	#   {
	#     domain: "domain.tld",
	#     users: [
	#       {
	#         email: "name@domain.tld",
	#         privileges: [ "priv1", "priv2", ... ],
	#         status: "active" | "inactive",
	#       },
	#       ...
	#     ]
	#   },
	#   ...
	# ]

	# Get users and their privileges.
	users = []
	active_accounts = set()
	c = open_database(env)
	c.execute('SELECT email, privileges FROM users')
	for email, privileges in c.fetchall():
		active_accounts.add(email)

		user = {
			"email": email,
			"privileges": parse_privs(privileges),
			"status": "active",
		}
		users.append(user)

	# Add in archived accounts.
	if with_archived:
		root = os.path.join(env['STORAGE_ROOT'], 'mail/mailboxes')
		for domain in os.listdir(root):
			if os.path.isdir(os.path.join(root, domain)):
				for user in os.listdir(os.path.join(root, domain)):
					email = user + "@" + domain
					mbox = os.path.join(root, domain, user)
					if email in active_accounts: continue
					user = {
						"email": email,
						"privileges": [],
						"status": "inactive",
						"mailbox": mbox,
					}
					users.append(user)

	# Group by domain.
	domains = { }
	for user in users:
		domain = get_domain(user["email"])
		if domain not in domains:
			domains[domain] = {
				"domain": domain,
				"users": []
				}
		domains[domain]["users"].append(user)

	# Sort domains.
	domains = [domains[domain] for domain in utils.sort_domains(domains.keys(), env)]

	# Sort users within each domain first by status then lexicographically by email address.
	for domain in domains:
		domain["users"].sort(key = lambda user : (user["status"] != "active", user["email"]))

	return domains

def get_admins(env):
	# Returns a set of users with admin privileges.
	users = set()
	for domain in get_mail_users_ex(env):
		for user in domain["users"]:
			if "admin" in user["privileges"]:
				users.add(user["email"])
	return users

def get_mail_aliases(env):
	# Returns a sorted list of tuples of (address, forward-tos, permitted-senders, auto).
	c = open_database(env)
	c.execute('SELECT source, destination, permitted_senders, 0 as auto FROM aliases UNION SELECT source, destination, permitted_senders, 1 as auto FROM auto_aliases')
	aliases = { row[0]: row for row in c.fetchall() } # make dict

	# put in a canonical order: sort by domain, then by email address lexicographically
	return [ aliases[address] for address in utils.sort_email_addresses(aliases.keys(), env) ]

def get_mail_aliases_ex(env):
	# Returns a complex data structure of all mail aliases, similar
	# to get_mail_users_ex.
	#
	# [
	#   {
	#     domain: "domain.tld",
	#     alias: [
	#       {
	#         address: "name@domain.tld", # IDNA-encoded
	#         address_display: "name@domain.tld", # full Unicode
	#         forwards_to: ["user1@domain.com", "receiver-only1@domain.com", ...],
	#         permitted_senders: ["user1@domain.com", "sender-only1@domain.com", ...] OR null,
	#         auto: True|False
	#       },
	#       ...
	#     ]
	#   },
	#   ...
	# ]

	domains = {}
	for address, forwards_to, permitted_senders, auto in get_mail_aliases(env):
		# skip auto domain maps since these are not informative in the control panel's aliases list
		if auto and address.startswith("@"): continue

		# get alias info
		domain = get_domain(address)

		# add to list
		if domain not in domains:
			domains[domain] = {
				"domain": domain,
				"aliases": [],
			}
		domains[domain]["aliases"].append({
			"address": address,
			"address_display": prettify_idn_email_address(address),
			"forwards_to": [prettify_idn_email_address(r.strip()) for r in forwards_to.split(",")],
			"permitted_senders": [prettify_idn_email_address(s.strip()) for s in permitted_senders.split(",")] if permitted_senders is not None else None,
			"auto": bool(auto),
		})

	# Sort domains.
	domains = [domains[domain] for domain in utils.sort_domains(domains.keys(), env)]

	# Sort aliases within each domain first by required-ness then lexicographically by address.
	for domain in domains:
		domain["aliases"].sort(key = lambda alias : (alias["auto"], alias["address"]))
	return domains

def get_domain(emailaddr, as_unicode=True):
	# Gets the domain part of an email address. Turns IDNA
	# back to Unicode for display.
	ret = emailaddr.split('@', 1)[1]
	if as_unicode:
		try:
			ret = idna.decode(ret.encode('ascii'))
		except (ValueError, UnicodeError, idna.IDNAError):
			# Looks like we have an invalid email address in
			# the database. Now is not the time to complain.
			pass
	return ret

def get_mail_domains(env, filter_aliases=lambda alias : True, users_only=False):
	# Returns the domain names (IDNA-encoded) of all of the email addresses
	# configured on the system. If users_only is True, only return domains
	# with email addresses that correspond to user accounts. Exclude Unicode
	# forms of domain names listed in the automatic aliases table.
	domains = []
	domains.extend([get_domain(login, as_unicode=False) for login in get_mail_users(env)])
	if not users_only:
		domains.extend([get_domain(address, as_unicode=False) for address, _, _, auto in get_mail_aliases(env) if filter_aliases(address) and not auto ])
	return set(domains)

def add_mail_user(email, pw, privs, env):
	# validate email
	if email.strip() == "":
		return ("No email address provided.", 400)
	elif not validate_email(email):
		return ("Invalid email address.", 400)
	elif not validate_email(email, mode='user'):
		return ("User account email addresses may only use the lowercase ASCII letters a-z, the digits 0-9, underscore (_), hyphen (-), and period (.).", 400)
	elif is_dcv_address(email) and len(get_mail_users(env)) > 0:
		# Make domain control validation hijacking a little harder to mess up by preventing the usual
		# addresses used for DCV from being user accounts. Except let it be the first account because
		# during box setup the user won't know the rules.
		return ("You may not make a user account for that address because it is frequently used for domain control validation. Use an alias instead if necessary.", 400)

	# validate password
	validate_password(pw)

	# validate privileges
	if privs is None or privs.strip() == "":
		privs = []
	else:
		privs = privs.split("\n")
		for p in privs:
			validation = validate_privilege(p)
			if validation: return validation

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

	# hash the password
	pw = hash_password(pw)

	# add the user to the database
	try:
		c.execute("INSERT INTO users (email, password, privileges) VALUES (?, ?, ?)",
			(email, pw, "\n".join(privs)))
	except sqlite3.IntegrityError:
		return ("User already exists.", 400)

	# write databasebefore next step
	conn.commit()

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

def set_mail_password(email, pw, env):
	# validate that password is acceptable
	validate_password(pw)

	# hash the password
	pw = hash_password(pw)

	# 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 hash_password(pw):
	# Turn the plain password into a Dovecot-format hashed password, meaning
	# something like "{SCHEME}hashedpassworddata".
	# http://wiki2.dovecot.org/Authentication/PasswordSchemes
	return utils.shell('check_output', ["/usr/bin/doveadm", "pw", "-s", "SHA512-CRYPT", "-p", pw]).strip()

def get_mail_password(email, env):
	# Gets the hashed password for a user. Passwords are stored in Dovecot's
	# password format, with a prefixed scheme.
	# http://wiki2.dovecot.org/Authentication/PasswordSchemes
	# update the database
	c = open_database(env)
	c.execute('SELECT password FROM users WHERE email=?', (email,))
	rows = c.fetchall()
	if len(rows) != 1:
		raise ValueError("That's not a user (%s)." % email)
	return rows[0][0]

def remove_mail_user(email, env):
	# remove
	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 things in case any domains are removed.
	return kick(env, "mail user removed")

def parse_privs(value):
	return [p for p in value.split("\n") if p.strip() != ""]

def get_mail_user_privileges(email, env, empty_on_error=False):
	# get privs
	c = open_database(env)
	c.execute('SELECT privileges FROM users WHERE email=?', (email,))
	rows = c.fetchall()
	if len(rows) != 1:
		if empty_on_error: return []
		return ("That's not a user (%s)." % email, 400)
	return parse_privs(rows[0][0])

def validate_privilege(priv):
	if "\n" in priv or priv.strip() == "":
		return ("That's not a valid privilege (%s)." % priv, 400)
	return None

def add_remove_mail_user_privilege(email, priv, action, env):
	# validate
	validation = validate_privilege(priv)
	if validation: return validation

	# get existing privs, but may fail
	privs = get_mail_user_privileges(email, env)
	if isinstance(privs, tuple): return privs # error

	# update privs set
	if action == "add":
		if priv not in privs:
			privs.append(priv)
	elif action == "remove":
		privs = [p for p in privs if p != priv]
	else:
		return ("Invalid action.", 400)

	# commit to database
	conn, c = open_database(env, with_connection=True)
	c.execute("UPDATE users SET privileges=? WHERE email=?", ("\n".join(privs), email))
	if c.rowcount != 1:
		return ("Something went wrong.", 400)
	conn.commit()

	return "OK"

def add_mail_alias(address, forwards_to, permitted_senders, env, update_if_exists=False, do_kick=True):
	# convert Unicode domain to IDNA
	address = sanitize_idn_email_address(address)

	# Our database is case sensitive (oops), which affects mail delivery
	# (Postfix always queries in lowercase?), so force lowercase.
	address = address.lower()

	# validate address
	address = address.strip()
	if address == "":
		return ("No email address provided.", 400)
	if not validate_email(address, mode='alias'):
		return ("Invalid email address (%s)." % address, 400)

	# validate forwards_to
	validated_forwards_to = []
	forwards_to = forwards_to.strip()

	# extra checks for email addresses used in domain control validation
	is_dcv_source = is_dcv_address(address)

	# Postfix allows a single @domain.tld as the destination, which means
	# the local part on the address is preserved in the rewrite. We must
	# try to convert Unicode to IDNA first before validating that it's a
	# legitimate alias address. Don't allow this sort of rewriting for
	# DCV source addresses.
	r1 = sanitize_idn_email_address(forwards_to)
	if validate_email(r1, mode='alias') and not is_dcv_source:
		validated_forwards_to.append(r1)

	else:
		# Parse comma and \n-separated destination emails & validate. In this
		# case, the forwards_to must be complete email addresses.
		for line in forwards_to.split("\n"):
			for email in line.split(","):
				email = email.strip()
				if email == "": continue
				email = sanitize_idn_email_address(email) # Unicode => IDNA
				# Strip any +tag from email alias and check privileges
				privileged_email = re.sub(r"(?=\+)[^@]*(?=@)",'',email)
				if not validate_email(email):
					return ("Invalid receiver email address (%s)." % email, 400)
				if is_dcv_source and not is_dcv_address(email) and "admin" not in get_mail_user_privileges(privileged_email, env, empty_on_error=True):
					# Make domain control validation hijacking a little harder to mess up by
					# requiring aliases for email addresses typically used in DCV to forward
					# only to accounts that are administrators on this system.
					return ("This alias can only have administrators of this system as destinations because the address is frequently used for domain control validation.", 400)
				validated_forwards_to.append(email)

	# validate permitted_senders
	valid_logins = get_mail_users(env)
	validated_permitted_senders = []
	permitted_senders = permitted_senders.strip()

	# Parse comma and \n-separated sender logins & validate. The permitted_senders must be
	# valid usernames.
	for line in permitted_senders.split("\n"):
		for login in line.split(","):
			login = login.strip()
			if login == "": continue
			if login not in valid_logins:
				return ("Invalid permitted sender: %s is not a user on this system." % login, 400)
			validated_permitted_senders.append(login)

	# Make sure the alias has either a forwards_to or a permitted_sender.
	if len(validated_forwards_to) + len(validated_permitted_senders) == 0:
		return ("The alias must either forward to an address or have a permitted sender.", 400)

	# save to db

	forwards_to = ",".join(validated_forwards_to)

	permitted_senders = None if len(validated_permitted_senders) == 0 else ",".join(validated_permitted_senders)

	conn, c = open_database(env, with_connection=True)
	try:
		c.execute("INSERT INTO aliases (source, destination, permitted_senders) VALUES (?, ?, ?)", (address, forwards_to, permitted_senders))
		return_status = "alias added"
	except sqlite3.IntegrityError:
		if not update_if_exists:
			return ("Alias already exists (%s)." % address, 400)
		else:
			c.execute("UPDATE aliases SET destination = ?, permitted_senders = ? WHERE source = ?", (forwards_to, permitted_senders, address))
			return_status = "alias updated"

	conn.commit()

	if do_kick:
		# Update things in case any new domains are added.
		return kick(env, return_status)
	return None

def remove_mail_alias(address, env, do_kick=True):
	# convert Unicode domain to IDNA
	address = sanitize_idn_email_address(address)

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

	if do_kick:
		# Update things in case any domains are removed.
		return kick(env, "alias removed")
	return None

def add_auto_aliases(aliases, env):
	conn, c = open_database(env, with_connection=True)
	c.execute("DELETE FROM auto_aliases")
	for source, destination in aliases.items():
		c.execute("INSERT INTO auto_aliases (source, destination) VALUES (?, ?)", (source, destination))
	conn.commit()

def get_system_administrator(env):
	return "administrator@" + env['PRIMARY_HOSTNAME']

def get_required_aliases(env):
	# These are the aliases that must exist.
	aliases = set()

	# The system administrator alias is required.
	aliases.add(get_system_administrator(env))

	# The hostmaster alias is exposed in the DNS SOA for each zone.
	aliases.add("hostmaster@" + env['PRIMARY_HOSTNAME'])

	# Get a list of domains we serve mail for, except ones for which the only
	# email on that domain are the required aliases or a catch-all/domain-forwarder.
	real_mail_domains = get_mail_domains(env,
		filter_aliases = lambda alias :
			not alias.startswith("postmaster@")
			and not alias.startswith("admin@")
			and not alias.startswith("abuse@")
			and not alias.startswith("@")
			)

	# Create postmaster@, admin@ and abuse@ for all domains we serve
	# mail on. postmaster@ is assumed to exist by our Postfix configuration.
	# admin@isn't anything, but it might save the user some trouble e.g. when
	# buying an SSL certificate.
	# abuse@ is part of RFC2142: https://www.ietf.org/rfc/rfc2142.txt
	for domain in real_mail_domains:
		aliases.add("postmaster@" + domain)
		aliases.add("admin@" + domain)
		aliases.add("abuse@" + domain)

	return aliases

def kick(env, mail_result=None):
	results = []

	# Include the current operation's result in output.

	if mail_result is not None:
		results.append(mail_result + "\n")

	auto_aliases = { }

	# Map required aliases to the administrator alias (which should be created manually).
	administrator = get_system_administrator(env)
	required_aliases = get_required_aliases(env)
	for alias in required_aliases:
		if alias == administrator: continue # don't make an alias from the administrator to itself --- this alias must be created manually
		auto_aliases[alias] = administrator

	# Add domain maps from Unicode forms of IDNA domains to the ASCII forms stored in the alias table.
	for domain in get_mail_domains(env):
		try:
			domain_unicode = idna.decode(domain.encode("ascii"))
			if domain == domain_unicode: continue # not an IDNA/Unicode domain
			auto_aliases["@" + domain_unicode] = "@" + domain
		except (ValueError, UnicodeError, idna.IDNAError):
			continue

	add_auto_aliases(auto_aliases, env)

	# Remove auto-generated postmaster/admin/abuse alises from the main aliases table.
	# They are now stored in the auto_aliases table.
	for address, forwards_to, _permitted_senders, auto in get_mail_aliases(env):
		user, domain = address.split("@")
		if user in {"postmaster", "admin", "abuse"} \
			and address not in required_aliases \
			and forwards_to == get_system_administrator(env) \
			and not auto:
			remove_mail_alias(address, env, do_kick=False)
			results.append(f"removed alias {address} (was to {forwards_to}; domain no longer used for email)\n")

	# Update DNS and nginx in case any domains are added/removed.

	from dns_update import do_dns_update
	results.append( do_dns_update(env) )

	from web_update import do_web_update
	results.append( do_web_update(env) )

	return "".join(s for s in results if s != "")

def validate_password(pw):
	# validate password
	if pw.strip() == "":
		msg = "No password provided."
		raise ValueError(msg)
	if len(pw) < 8:
		msg = "Passwords must be at least eight characters."
		raise ValueError(msg)

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], mode='user'):
			sys.exit(0)
		else:
			sys.exit(1)

	if len(sys.argv) > 1 and sys.argv[1] == "update":
		from utils import load_environment
		print(kick(load_environment()))



File: management/daemon.py

#!/usr/local/lib/mailinabox/env/bin/python3
#
# The API can be accessed on the command line, e.g. use `curl` like so:
#    curl --user $(</var/lib/mailinabox/api.key): http://localhost:10222/mail/users
#
# During development, you can start the Mail-in-a-Box control panel
# by running this script, e.g.:
#
# service mailinabox stop # stop the system process
# DEBUG=1 management/daemon.py
# service mailinabox start # when done debugging, start it up again

import os, os.path, re, json, time
import multiprocessing.pool

from functools import wraps

from flask import Flask, request, render_template, Response, send_from_directory, make_response

import auth, utils
from mailconfig import get_mail_users, get_mail_users_ex, get_admins, add_mail_user, set_mail_password, remove_mail_user
from mailconfig import get_mail_user_privileges, add_remove_mail_user_privilege
from mailconfig import get_mail_aliases, get_mail_aliases_ex, get_mail_domains, add_mail_alias, remove_mail_alias
from mfa import get_public_mfa_state, provision_totp, validate_totp_secret, enable_mfa, disable_mfa
import contextlib

env = utils.load_environment()

auth_service = auth.AuthService()

# We may deploy via a symbolic link, which confuses flask's template finding.
me = __file__
with contextlib.suppress(OSError):
	me = os.readlink(__file__)

# for generating CSRs we need a list of country codes
csr_country_codes = []
with open(os.path.join(os.path.dirname(me), "csr_country_codes.tsv"), encoding="utf-8") as f:
	for line in f:
		if line.strip() == "" or line.startswith("#"): continue
		code, name = line.strip().split("\t")[0:2]
		csr_country_codes.append((code, name))

app = Flask(__name__, template_folder=os.path.abspath(os.path.join(os.path.dirname(me), "templates")))

# Decorator to protect views that require a user with 'admin' privileges.
def authorized_personnel_only(viewfunc):
	@wraps(viewfunc)
	def newview(*args, **kwargs):
		# Authenticate the passed credentials, which is either the API key or a username:password pair
		# and an optional X-Auth-Token token.
		error = None
		privs = []

		try:
			email, privs = auth_service.authenticate(request, env)
		except ValueError as e:
			# Write a line in the log recording the failed login, unless no authorization header
			# was given which can happen on an initial request before a 403 response.
			if "Authorization" in request.headers:
				log_failed_login(request)

			# Authentication failed.
			error = str(e)

		# Authorized to access an API view?
		if "admin" in privs:
			# Store the email address of the logged in user so it can be accessed
			# from the API methods that affect the calling user.
			request.user_email = email
			request.user_privs = privs

			# Call view func.
			return viewfunc(*args, **kwargs)

		if not error:
			error = "You are not an administrator."

		# Not authorized. Return a 401 (send auth) and a prompt to authorize by default.
		status = 401
		headers = {
			'WWW-Authenticate': f'Basic realm="{auth_service.auth_realm}"',
			'X-Reason': error,
		}

		if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
			# Don't issue a 401 to an AJAX request because the user will
			# be prompted for credentials, which is not helpful.
			status = 403
			headers = None

		if request.headers.get('Accept') in {None, "", "*/*"}:
			# Return plain text output.
			return Response(error+"\n", status=status, mimetype='text/plain', headers=headers)
		else:
			# Return JSON output.
			return Response(json.dumps({
				"status": "error",
				"reason": error,
				})+"\n", status=status, mimetype='application/json', headers=headers)

	return newview

@app.errorhandler(401)
def unauthorized(error):
	return auth_service.make_unauthorized_response()

def json_response(data, status=200):
	return Response(json.dumps(data, indent=2, sort_keys=True)+'\n', status=status, mimetype='application/json')

###################################

# Control Panel (unauthenticated views)

@app.route('/')
def index():
	# Render the control panel. This route does not require user authentication
	# so it must be safe!

	no_users_exist = (len(get_mail_users(env)) == 0)
	no_admins_exist = (len(get_admins(env)) == 0)

	import boto3.s3
	backup_s3_hosts = [(r, f"s3.{r}.amazonaws.com") for r in boto3.session.Session().get_available_regions('s3')]


	return render_template('index.html',
		hostname=env['PRIMARY_HOSTNAME'],
		storage_root=env['STORAGE_ROOT'],

		no_users_exist=no_users_exist,
		no_admins_exist=no_admins_exist,

		backup_s3_hosts=backup_s3_hosts,
		csr_country_codes=csr_country_codes,
	)

# Create a session key by checking the username/password in the Authorization header.
@app.route('/login', methods=["POST"])
def login():
	# Is the caller authorized?
	try:
		email, privs = auth_service.authenticate(request, env, login_only=True)
	except ValueError as e:
		if "missing-totp-token" in str(e):
			return json_response({
				"status": "missing-totp-token",
				"reason": str(e),
			})
		else:
			# Log the failed login
			log_failed_login(request)
			return json_response({
				"status": "invalid",
				"reason": str(e),
			})

	# Return a new session for the user.
	resp = {
		"status": "ok",
		"email": email,
		"privileges": privs,
		"api_key": auth_service.create_session_key(email, env, type='login'),
	}

	app.logger.info(f"New login session created for {email}")

	# Return.
	return json_response(resp)

@app.route('/logout', methods=["POST"])
def logout():
	try:
		email, _ = auth_service.authenticate(request, env, logout=True)
		app.logger.info(f"{email} logged out")
	except ValueError:
		pass
	finally:
		return json_response({ "status": "ok" })

# MAIL

@app.route('/mail/users')
@authorized_personnel_only
def mail_users():
	if request.args.get("format", "") == "json":
		return json_response(get_mail_users_ex(env, with_archived=True))
	else:
		return "".join(x+"\n" for x in get_mail_users(env))

@app.route('/mail/users/add', methods=['POST'])
@authorized_personnel_only
def mail_users_add():
	try:
		return add_mail_user(request.form.get('email', ''), request.form.get('password', ''), request.form.get('privileges', ''), env)
	except ValueError as e:
		return (str(e), 400)

@app.route('/mail/users/password', methods=['POST'])
@authorized_personnel_only
def mail_users_password():
	try:
		return set_mail_password(request.form.get('email', ''), request.form.get('password', ''), env)
	except ValueError as e:
		return (str(e), 400)

@app.route('/mail/users/remove', methods=['POST'])
@authorized_personnel_only
def mail_users_remove():
	return remove_mail_user(request.form.get('email', ''), env)


@app.route('/mail/users/privileges')
@authorized_personnel_only
def mail_user_privs():
	privs = get_mail_user_privileges(request.args.get('email', ''), env)
	if isinstance(privs, tuple): return privs # error
	return "\n".join(privs)

@app.route('/mail/users/privileges/add', methods=['POST'])
@authorized_personnel_only
def mail_user_privs_add():
	return add_remove_mail_user_privilege(request.form.get('email', ''), request.form.get('privilege', ''), "add", env)

@app.route('/mail/users/privileges/remove', methods=['POST'])
@authorized_personnel_only
def mail_user_privs_remove():
	return add_remove_mail_user_privilege(request.form.get('email', ''), request.form.get('privilege', ''), "remove", env)


@app.route('/mail/aliases')
@authorized_personnel_only
def mail_aliases():
	if request.args.get("format", "") == "json":
		return json_response(get_mail_aliases_ex(env))
	else:
		return "".join(address+"\t"+receivers+"\t"+(senders or "")+"\n" for address, receivers, senders, auto in get_mail_aliases(env))

@app.route('/mail/aliases/add', methods=['POST'])
@authorized_personnel_only
def mail_aliases_add():
	return add_mail_alias(
		request.form.get('address', ''),
		request.form.get('forwards_to', ''),
		request.form.get('permitted_senders', ''),
		env,
		update_if_exists=(request.form.get('update_if_exists', '') == '1')
		)

@app.route('/mail/aliases/remove', methods=['POST'])
@authorized_personnel_only
def mail_aliases_remove():
	return remove_mail_alias(request.form.get('address', ''), env)

@app.route('/mail/domains')
@authorized_personnel_only
def mail_domains():
    return "".join(x+"\n" for x in get_mail_domains(env))

# DNS

@app.route('/dns/zones')
@authorized_personnel_only
def dns_zones():
	from dns_update import get_dns_zones
	return json_response([z[0] for z in get_dns_zones(env)])

@app.route('/dns/update', methods=['POST'])
@authorized_personnel_only
def dns_update():
	from dns_update import do_dns_update
	try:
		return do_dns_update(env, force=request.form.get('force', '') == '1')
	except Exception as e:
		return (str(e), 500)

@app.route('/dns/secondary-nameserver')
@authorized_personnel_only
def dns_get_secondary_nameserver():
	from dns_update import get_custom_dns_config, get_secondary_dns
	return json_response({ "hostnames": get_secondary_dns(get_custom_dns_config(env), mode=None) })

@app.route('/dns/secondary-nameserver', methods=['POST'])
@authorized_personnel_only
def dns_set_secondary_nameserver():
	from dns_update import set_secondary_dns
	try:
		return set_secondary_dns([ns.strip() for ns in re.split(r"[, ]+", request.form.get('hostnames') or "") if ns.strip() != ""], env)
	except ValueError as e:
		return (str(e), 400)

@app.route('/dns/custom')
@authorized_personnel_only
def dns_get_records(qname=None, rtype=None):
	# Get the current set of custom DNS records.
	from dns_update import get_custom_dns_config, get_dns_zones
	records = get_custom_dns_config(env, only_real_records=True)

	# Filter per the arguments for the more complex GET routes below.
	records = [r for r in records
		if (not qname or r[0] == qname)
		and (not rtype or r[1] == rtype) ]

	# Make a better data structure.
	records = [
        {
                "qname": r[0],
                "rtype": r[1],
                "value": r[2],
		"sort-order": { },
        } for r in records ]

	# To help with grouping by zone in qname sorting, label each record with which zone it is in.
	# There's an inconsistency in how we handle zones in get_dns_zones and in sort_domains, so
	# do this first before sorting the domains within the zones.
	zones = utils.sort_domains([z[0] for z in get_dns_zones(env)], env)
	for r in records:
		for z in zones:
			if r["qname"] == z or r["qname"].endswith("." + z):
				r["zone"] = z
				break

	# Add sorting information. The 'created' order follows the order in the YAML file on disk,
	# which tracs the order entries were added in the control panel since we append to the end.
	# The 'qname' sort order sorts by our standard domain name sort (by zone then by qname),
	# then by rtype, and last by the original order in the YAML file (since sorting by value
	# may not make sense, unless we parse IP addresses, for example).
	for i, r in enumerate(records):
		r["sort-order"]["created"] = i
	domain_sort_order = utils.sort_domains([r["qname"] for r in records], env)
	for i, r in enumerate(sorted(records, key = lambda r : (
			zones.index(r["zone"]) if r.get("zone") else 0, # record is not within a zone managed by the box
			domain_sort_order.index(r["qname"]),
			r["rtype"]))):
		r["sort-order"]["qname"] = i

	# Return.
	return json_response(records)

@app.route('/dns/custom/<qname>', methods=['GET', 'POST', 'PUT', 'DELETE'])
@app.route('/dns/custom/<qname>/<rtype>', methods=['GET', 'POST', 'PUT', 'DELETE'])
@authorized_personnel_only
def dns_set_record(qname, rtype="A"):
	from dns_update import do_dns_update, set_custom_dns_record
	try:
		# Normalize.
		rtype = rtype.upper()

		# Read the record value from the request BODY, which must be
		# ASCII-only. Not used with GET.
		value = request.stream.read().decode("ascii", "ignore").strip()

		if request.method == "GET":
			# Get the existing records matching the qname and rtype.
			return dns_get_records(qname, rtype)

		elif request.method in {"POST", "PUT"}:
			# There is a default value for A/AAAA records.
			if rtype in {"A", "AAAA"} and value == "":
				value = request.environ.get("HTTP_X_FORWARDED_FOR") # normally REMOTE_ADDR but we're behind nginx as a reverse proxy

			# Cannot add empty records.
			if value == '':
				return ("No value for the record provided.", 400)

			if request.method == "POST":
				# Add a new record (in addition to any existing records
				# for this qname-rtype pair).
				action = "add"
			elif request.method == "PUT":
				# In REST, PUT is supposed to be idempotent, so we'll
				# make this action set (replace all records for this
				# qname-rtype pair) rather than add (add a new record).
				action = "set"

		elif request.method == "DELETE":
			if value == '':
				# Delete all records for this qname-type pair.
				value = None
			else:
				# Delete just the qname-rtype-value record exactly.
				pass
			action = "remove"

		if set_custom_dns_record(qname, rtype, value, action, env):
			return do_dns_update(env) or "Something isn't right."
		return "OK"

	except ValueError as e:
		return (str(e), 400)

@app.route('/dns/dump')
@authorized_personnel_only
def dns_get_dump():
	from dns_update import build_recommended_dns
	return json_response(build_recommended_dns(env))

@app.route('/dns/zonefile/<zone>')
@authorized_personnel_only
def dns_get_zonefile(zone):
	from dns_update import get_dns_zonefile
	return Response(get_dns_zonefile(zone, env), status=200, mimetype='text/plain')

# SSL

@app.route('/ssl/status')
@authorized_personnel_only
def ssl_get_status():
	from ssl_certificates import get_certificates_to_provision
	from web_update import get_web_domains_info, get_web_domains

	# What domains can we provision certificates for? What unexpected problems do we have?
	provision, cant_provision = get_certificates_to_provision(env, show_valid_certs=False)

	# What's the current status of TLS certificates on all of the domain?
	domains_status = get_web_domains_info(env)
	domains_status = [
		{
			"domain": d["domain"],
			"status": d["ssl_certificate"][0],
			"text": d["ssl_certificate"][1] + (" " + cant_provision[d["domain"]] if d["domain"] in cant_provision else "")
		} for d in domains_status ]

	# Warn the user about domain names not hosted here because of other settings.
	for domain in set(get_web_domains(env, exclude_dns_elsewhere=False)) - set(get_web_domains(env)):
		domains_status.append({
			"domain": domain,
			"status": "not-applicable",
			"text": "The domain's website is hosted elsewhere.",
		})

	return json_response({
		"can_provision": utils.sort_domains(provision, env),
		"status": domains_status,
	})

@app.route('/ssl/csr/<domain>', methods=['POST'])
@authorized_personnel_only
def ssl_get_csr(domain):
	from ssl_certificates import create_csr
	ssl_private_key = os.path.join(os.path.join(env["STORAGE_ROOT"], 'ssl', 'ssl_private_key.pem'))
	return create_csr(domain, ssl_private_key, request.form.get('countrycode', ''), env)

@app.route('/ssl/install', methods=['POST'])
@authorized_personnel_only
def ssl_install_cert():
	from web_update import get_web_domains
	from ssl_certificates import install_cert
	domain = request.form.get('domain')
	ssl_cert = request.form.get('cert')
	ssl_chain = request.form.get('chain')
	if domain not in get_web_domains(env):
		return "Invalid domain name."
	return install_cert(domain, ssl_cert, ssl_chain, env)

@app.route('/ssl/provision', methods=['POST'])
@authorized_personnel_only
def ssl_provision_certs():
	from ssl_certificates import provision_certificates
	requests = provision_certificates(env, limit_domains=None)
	return json_response({ "requests": requests })

# multi-factor auth

@app.route('/mfa/status', methods=['POST'])
@authorized_personnel_only
def mfa_get_status():
	# Anyone accessing this route is an admin, and we permit them to
	# see the MFA status for any user if they submit a 'user' form
	# field. But we don't include provisioning info since a user can
	# only provision for themselves.
	email = request.form.get('user', request.user_email) # user field if given, otherwise the user making the request
	try:
		resp = {
			"enabled_mfa": get_public_mfa_state(email, env)
		}
		if email == request.user_email:
			resp.update({
				"new_mfa": {
					"totp": provision_totp(email, env)
				}
			})
	except ValueError as e:
		return (str(e), 400)
	return json_response(resp)

@app.route('/mfa/totp/enable', methods=['POST'])
@authorized_personnel_only
def totp_post_enable():
	secret = request.form.get('secret')
	token = request.form.get('token')
	label = request.form.get('label')
	if not isinstance(token, str):
		return ("Bad Input", 400)
	try:
		validate_totp_secret(secret)
		enable_mfa(request.user_email, "totp", secret, token, label, env)
	except ValueError as e:
		return (str(e), 400)
	return "OK"

@app.route('/mfa/disable', methods=['POST'])
@authorized_personnel_only
def totp_post_disable():
	# Anyone accessing this route is an admin, and we permit them to
	# disable the MFA status for any user if they submit a 'user' form
	# field.
	email = request.form.get('user', request.user_email) # user field if given, otherwise the user making the request
	try:
		result = disable_mfa(email, request.form.get('mfa-id') or None, env) # convert empty string to None
	except ValueError as e:
		return (str(e), 400)
	if result: # success
		return "OK"
	else: # error
		return ("Invalid user or MFA id.", 400)

# WEB

@app.route('/web/domains')
@authorized_personnel_only
def web_get_domains():
	from web_update import get_web_domains_info
	return json_response(get_web_domains_info(env))

@app.route('/web/update', methods=['POST'])
@authorized_personnel_only
def web_update():
	from web_update import do_web_update
	return do_web_update(env)

# System

@app.route('/system/version', methods=["GET"])
@authorized_personnel_only
def system_version():
	from status_checks import what_version_is_this
	try:
		return what_version_is_this(env)
	except Exception as e:
		return (str(e), 500)

@app.route('/system/latest-upstream-version', methods=["POST"])
@authorized_personnel_only
def system_latest_upstream_version():
	from status_checks import get_latest_miab_version
	try:
		return get_latest_miab_version()
	except Exception as e:
		return (str(e), 500)

@app.route('/system/status', methods=["POST"])
@authorized_personnel_only
def system_status():
	from status_checks import run_checks
	class WebOutput:
		def __init__(self):
			self.items = []
		def add_heading(self, heading):
			self.items.append({ "type": "heading", "text": heading, "extra": [] })
		def print_ok(self, message):
			self.items.append({ "type": "ok", "text": message, "extra": [] })
		def print_error(self, message):
			self.items.append({ "type": "error", "text": message, "extra": [] })
		def print_warning(self, message):
			self.items.append({ "type": "warning", "text": message, "extra": [] })
		def print_line(self, message, monospace=False):
			self.items[-1]["extra"].append({ "text": message, "monospace": monospace })
	output = WebOutput()
	# Create a temporary pool of processes for the status checks
	with multiprocessing.pool.Pool(processes=5) as pool:
		run_checks(False, env, output, pool)
		pool.close()
		pool.join()
	return json_response(output.items)

@app.route('/system/updates')
@authorized_personnel_only
def show_updates():
	from status_checks import list_apt_updates
	return "".join(
		"{} ({})\n".format(p["package"], p["version"])
		for p in list_apt_updates())

@app.route('/system/update-packages', methods=["POST"])
@authorized_personnel_only
def do_updates():
	utils.shell("check_call", ["/usr/bin/apt-get", "-qq", "update"])
	return utils.shell("check_output", ["/usr/bin/apt-get", "-y", "upgrade"], env={
		"DEBIAN_FRONTEND": "noninteractive"
	})


@app.route('/system/reboot', methods=["GET"])
@authorized_personnel_only
def needs_reboot():
	from status_checks import is_reboot_needed_due_to_package_installation
	if is_reboot_needed_due_to_package_installation():
		return json_response(True)
	else:
		return json_response(False)

@app.route('/system/reboot', methods=["POST"])
@authorized_personnel_only
def do_reboot():
	# To keep the attack surface low, we don't allow a remote reboot if one isn't necessary.
	from status_checks import is_reboot_needed_due_to_package_installation
	if is_reboot_needed_due_to_package_installation():
		return utils.shell("check_output", ["/sbin/shutdown", "-r", "now"], capture_stderr=True)
	else:
		return "No reboot is required, so it is not allowed."


@app.route('/system/backup/status')
@authorized_personnel_only
def backup_status():
	from backup import backup_status
	try:
		return json_response(backup_status(env))
	except Exception as e:
		return json_response({ "error": str(e) })

@app.route('/system/backup/config', methods=["GET"])
@authorized_personnel_only
def backup_get_custom():
	from backup import get_backup_config
	return json_response(get_backup_config(env, for_ui=True))

@app.route('/system/backup/config', methods=["POST"])
@authorized_personnel_only
def backup_set_custom():
	from backup import backup_set_custom
	return json_response(backup_set_custom(env,
		request.form.get('target', ''),
		request.form.get('target_user', ''),
		request.form.get('target_pass', ''),
		request.form.get('min_age', '')
	))

@app.route('/system/privacy', methods=["GET"])
@authorized_personnel_only
def privacy_status_get():
	config = utils.load_settings(env)
	return json_response(config.get("privacy", True))

@app.route('/system/privacy', methods=["POST"])
@authorized_personnel_only
def privacy_status_set():
	config = utils.load_settings(env)
	config["privacy"] = (request.form.get('value') == "private")
	utils.write_settings(config, env)
	return "OK"

# MUNIN

@app.route('/munin/')
@authorized_personnel_only
def munin_start():
	# Munin pages, static images, and dynamically generated images are served
	# outside of the AJAX API. We'll start with a 'start' API that sets a cookie
	# that subsequent requests will read for authorization. (We don't use cookies
	# for the API to avoid CSRF vulnerabilities.)
	response = make_response("OK")
	response.set_cookie("session", auth_service.create_session_key(request.user_email, env, type='cookie'),
	    max_age=60*30, secure=True, httponly=True, samesite="Strict") # 30 minute duration
	return response

def check_request_cookie_for_admin_access():
	session = auth_service.get_session(None, request.cookies.get("session", ""), "cookie", env)
	if not session: return False
	privs = get_mail_user_privileges(session["email"], env)
	if not isinstance(privs, list): return False
	if "admin" not in privs: return False
	return True

def authorized_personnel_only_via_cookie(f):
	@wraps(f)
	def g(*args, **kwargs):
		if not check_request_cookie_for_admin_access():
			return Response("Unauthorized", status=403, mimetype='text/plain', headers={})
		return f(*args, **kwargs)
	return g

@app.route('/munin/<path:filename>')
@authorized_personnel_only_via_cookie
def munin_static_file(filename=""):
	# Proxy the request to static files.
	if filename == "": filename = "index.html"
	return send_from_directory("/var/cache/munin/www", filename)

@app.route('/munin/cgi-graph/<path:filename>')
@authorized_personnel_only_via_cookie
def munin_cgi(filename):
	""" Relay munin cgi dynazoom requests
	/usr/lib/munin/cgi/munin-cgi-graph is a perl cgi script in the munin package
	that is responsible for generating binary png images _and_ associated HTTP
	headers based on parameters in the requesting URL. All output is written
	to stdout which munin_cgi splits into response headers and binary response
	data.
	munin-cgi-graph reads environment variables to determine
	what it should do. It expects a path to be in the env-var PATH_INFO, and a
	querystring to be in the env-var QUERY_STRING.
	munin-cgi-graph has several failure modes. Some write HTTP Status headers and
	others return nonzero exit codes.
	Situating munin_cgi between the user-agent and munin-cgi-graph enables keeping
	the cgi script behind mailinabox's auth mechanisms and avoids additional
	support infrastructure like spawn-fcgi.
	"""

	COMMAND = 'su munin --preserve-environment --shell=/bin/bash -c /usr/lib/munin/cgi/munin-cgi-graph'
	# su changes user, we use the munin user here
	# --preserve-environment retains the environment, which is where Popen's `env` data is
	# --shell=/bin/bash ensures the shell used is bash
	# -c "/usr/lib/munin/cgi/munin-cgi-graph" passes the command to run as munin
	# "%s" is a placeholder for where the request's querystring will be added

	if filename == "":
		return ("a path must be specified", 404)

	query_str = request.query_string.decode("utf-8", 'ignore')

	env = {'PATH_INFO': '/%s/' % filename, 'REQUEST_METHOD': 'GET', 'QUERY_STRING': query_str}
	code, binout = utils.shell('check_output',
							   COMMAND.split(" ", 5),
							   # Using a maxsplit of 5 keeps the last arguments together
							   env=env,
							   return_bytes=True,
							   trap=True)

	if code != 0:
		# nonzero returncode indicates error
		app.logger.error("munin_cgi: munin-cgi-graph returned nonzero exit code, %s", code)
		return ("error processing graph image", 500)

	# /usr/lib/munin/cgi/munin-cgi-graph returns both headers and binary png when successful.
	# A double-Windows-style-newline always indicates the end of HTTP headers.
	headers, image_bytes = binout.split(b'\r\n\r\n', 1)
	response = make_response(image_bytes)
	for line in headers.splitlines():
		name, value = line.decode("utf8").split(':', 1)
		response.headers[name] = value
	if 'Status' in response.headers and '404' in response.headers['Status']:
		app.logger.warning("munin_cgi: munin-cgi-graph returned 404 status code. PATH_INFO=%s", env['PATH_INFO'])
	return response

def log_failed_login(request):
	# We need to figure out the ip to list in the message, all our calls are routed
	# through nginx who will put the original ip in X-Forwarded-For.
	# During setup we call the management interface directly to determine the user
	# status. So we can't always use X-Forwarded-For because during setup that header
	# will not be present.
	ip = request.headers.getlist("X-Forwarded-For")[0] if request.headers.getlist("X-Forwarded-For") else request.remote_addr

	# We need to add a timestamp to the log message, otherwise /dev/log will eat the "duplicate"
	# message.
	app.logger.warning( f"Mail-in-a-Box Management Daemon: Failed login attempt from ip {ip} - timestamp {time.time()}")


# APP

if __name__ == '__main__':
	if "DEBUG" in os.environ:
		# Turn on Flask debugging.
		app.debug = True

	if not app.debug:
		app.logger.addHandler(utils.create_syslog_handler())

	#app.logger.info('API key: ' + auth_service.key)

	# Start the application server. Listens on 127.0.0.1 (IPv4 only).
	app.run(port=10222)



File: management/wsgi.py

from daemon import app
import utils

app.logger.addHandler(utils.create_syslog_handler())

if __name__ == "__main__":
    app.run(port=10222)



Mail-in-a-Box
=============

By [@JoshData](https://github.com/JoshData) and [contributors](https://github.com/mail-in-a-box/mailinabox/graphs/contributors).

Mail-in-a-Box helps individuals take back control of their email by defining a one-click, easy-to-deploy SMTP+everything else server: a mail server in a box.

**Please see [https://mailinabox.email](https://mailinabox.email) for the project's website and setup guide!**

* * *

Our goals are to:

* Make deploying a good mail server easy.
* Promote [decentralization](http://redecentralize.org/), innovation, and privacy on the web.
* Have automated, auditable, and [idempotent](https://web.archive.org/web/20190518072631/https://sharknet.us/2014/02/01/automated-configuration-management-challenges-with-idempotency/) configuration.
* **Not** make a totally unhackable, NSA-proof server.
* **Not** make something customizable by power users.

Additionally, this project has a [Code of Conduct](CODE_OF_CONDUCT.md), which supersedes the goals above. Please review it when joining our community.


In The Box
----------

Mail-in-a-Box turns a fresh Ubuntu 22.04 LTS 64-bit machine into a working mail server by installing and configuring various components.

It is a one-click email appliance. There are no user-configurable setup options. It "just works."

The components installed are:

* SMTP ([postfix](http://www.postfix.org/)), IMAP ([Dovecot](http://dovecot.org/)), CardDAV/CalDAV ([Nextcloud](https://nextcloud.com/)), and Exchange ActiveSync ([z-push](http://z-push.org/)) servers
* Webmail ([Roundcube](http://roundcube.net/)), mail filter rules (thanks to Roundcube and Dovecot), and email client autoconfig settings (served by [nginx](http://nginx.org/))
* Spam filtering ([spamassassin](https://spamassassin.apache.org/)) and greylisting ([postgrey](http://postgrey.schweikert.ch/))
* DNS ([nsd4](https://www.nlnetlabs.nl/projects/nsd/)) with [SPF](https://en.wikipedia.org/wiki/Sender_Policy_Framework), DKIM ([OpenDKIM](http://www.opendkim.org/)), [DMARC](https://en.wikipedia.org/wiki/DMARC), [DNSSEC](https://en.wikipedia.org/wiki/DNSSEC), [DANE TLSA](https://en.wikipedia.org/wiki/DNS-based_Authentication_of_Named_Entities), [MTA-STS](https://tools.ietf.org/html/rfc8461), and [SSHFP](https://tools.ietf.org/html/rfc4255) policy records automatically set
* TLS certificates are automatically provisioned using [Let's Encrypt](https://letsencrypt.org/) for protecting https and all of the other services on the box
* Backups ([duplicity](http://duplicity.nongnu.org/)), firewall ([ufw](https://launchpad.net/ufw)), intrusion protection ([fail2ban](http://www.fail2ban.org/wiki/index.php/Main_Page)), and basic system monitoring ([munin](http://munin-monitoring.org/))

It also includes system management tools:

* Comprehensive health monitoring that checks each day that services are running, ports are open, TLS certificates are valid, and DNS records are correct
* A control panel for adding/removing mail users, aliases, custom DNS records, configuring backups, etc.
* An API for all of the actions on the control panel

Internationalized domain names are supported and configured easily (but SMTPUTF8 is not supported, unfortunately).

It also supports static website hosting since the box is serving HTTPS anyway. (To serve a website for your domains elsewhere, just add a custom DNS "A" record in you Mail-in-a-Box's control panel to point domains to another server.)

For more information on how Mail-in-a-Box handles your privacy, see the [security details page](security.md).


Installation
------------

See the [setup guide](https://mailinabox.email/guide.html) for detailed, user-friendly instructions.

For experts, start with a completely fresh (really, I mean it) Ubuntu 22.04 LTS 64-bit machine. On the machine...

Clone this repository and checkout the tag corresponding to the most recent release (which you can find in the tags or releases lists on GitHub):

	$ git clone https://github.com/mail-in-a-box/mailinabox
	$ cd mailinabox
	$ git checkout TAGNAME

Begin the installation.

	$ sudo setup/start.sh

The installation will install, uninstall, and configure packages to turn the machine into a working, good mail server.

For help, DO NOT contact Josh directly --- I don't do tech support by email or tweet (no exceptions).

Post your question on the [discussion forum](https://discourse.mailinabox.email/) instead, where maintainers and Mail-in-a-Box users may be able to help you.

Note that while we want everything to "just work," we can't control the rest of the Internet. Other mail services might block or spam-filter email sent from your Mail-in-a-Box.
This is a challenge faced by everyone who runs their own mail server, with or without Mail-in-a-Box. See our discussion forum for tips about that.


Contributing and Development
----------------------------

Mail-in-a-Box is an open source project. Your contributions and pull requests are welcome. See [CONTRIBUTING](CONTRIBUTING.md) to get started. 


The Acknowledgements
--------------------

This project was inspired in part by the ["NSA-proof your email in 2 hours"](http://sealedabstract.com/code/nsa-proof-your-e-mail-in-2-hours/) blog post by Drew Crawford, [Sovereign](https://github.com/sovereign/sovereign) by Alex Payne, and conversations with <a href="https://twitter.com/shevski" target="_blank">@shevski</a>, <a href="https://github.com/konklone" target="_blank">@konklone</a>, and <a href="https://github.com/gregelin" target="_blank">@GregElin</a>.

Mail-in-a-Box is similar to [iRedMail](http://www.iredmail.org/) and [Modoboa](https://github.com/tonioo/modoboa).


The History
-----------

* In 2007 I wrote a relatively popular Mozilla Thunderbird extension that added client-side SPF and DKIM checks to mail to warn users about possible phishing: [add-on page](https://addons.mozilla.org/en-us/thunderbird/addon/sender-verification-anti-phish/), [source](https://github.com/JoshData/thunderbird-spf).
* In August 2013 I began Mail-in-a-Box by combining my own mail server configuration with the setup in ["NSA-proof your email in 2 hours"](http://sealedabstract.com/code/nsa-proof-your-e-mail-in-2-hours/) and making the setup steps reproducible with bash scripts.
* Mail-in-a-Box was a semifinalist in the 2014 [Knight News Challenge](https://www.newschallenge.org/challenge/2014/submissions/mail-in-a-box), but it was not selected as a winner.
* Mail-in-a-Box hit the front page of Hacker News in [April](https://news.ycombinator.com/item?id=7634514) 2014, [September](https://news.ycombinator.com/item?id=8276171) 2014, [May](https://news.ycombinator.com/item?id=9624267) 2015, and [November](https://news.ycombinator.com/item?id=13050500) 2016.
* FastCompany mentioned Mail-in-a-Box a [roundup of privacy projects](http://www.fastcompany.com/3047645/your-own-private-cloud) on June 26, 2015.
