treewide: reformat python code

This commit is contained in:
Martin Weinelt 2025-05-08 01:40:37 +02:00
parent f9fcbe9430
commit a7d580b934
No known key found for this signature in database
GPG Key ID: 87C1E9888F856759
3 changed files with 185 additions and 129 deletions

View File

@ -17,9 +17,9 @@
# -- Project information -----------------------------------------------------
project = 'NixOS Mailserver'
copyright = '2022, NixOS Mailserver Contributors'
author = 'NixOS Mailserver Contributors'
project = "NixOS Mailserver"
copyright = "2022, NixOS Mailserver Contributors"
author = "NixOS Mailserver Contributors"
# -- General configuration ---------------------------------------------------
@ -27,33 +27,31 @@ author = 'NixOS Mailserver Contributors'
# Add any Sphinx extension module names here, as strings. They can be
# extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
# ones.
extensions = [
'myst_parser'
]
extensions = ["myst_parser"]
myst_enable_extensions = [
'colon_fence',
'linkify',
"colon_fence",
"linkify",
]
smartquotes = False
# Add any paths that contain templates here, relative to this directory.
templates_path = ['_templates']
templates_path = ["_templates"]
# List of patterns, relative to source directory, that match files and
# directories to ignore when looking for source files.
# This pattern also affects html_static_path and html_extra_path.
exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store']
exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"]
master_doc = 'index'
master_doc = "index"
# -- Options for HTML output -------------------------------------------------
# The theme to use for HTML and HTML Help pages. See the documentation for
# a list of builtin themes.
#
html_theme = 'sphinx_rtd_theme'
html_theme = "sphinx_rtd_theme"
# Add any paths that contain custom static files (such as style sheets) here,
# relative to this directory. They are copied after the builtin static files,

View File

@ -21,64 +21,71 @@ template = """
f = open(sys.argv[1])
options = json.load(f)
groups = ["mailserver.loginAccounts",
"mailserver.certificate",
"mailserver.dkim",
"mailserver.dmarcReporting",
"mailserver.fullTextSearch",
"mailserver.redis",
"mailserver.ldap",
"mailserver.monitoring",
"mailserver.backup",
"mailserver.borgbackup"]
groups = [
"mailserver.loginAccounts",
"mailserver.certificate",
"mailserver.dkim",
"mailserver.dmarcReporting",
"mailserver.fullTextSearch",
"mailserver.redis",
"mailserver.ldap",
"mailserver.monitoring",
"mailserver.backup",
"mailserver.borgbackup",
]
def render_option_value(opt, attr):
if attr not in opt:
return ""
if isinstance(opt[attr], dict) and '_type' in opt[attr]:
if opt[attr]['_type'] == 'literalExpression':
if '\n' in opt[attr]['text']:
res = '\n```nix\n' + opt[attr]['text'].rstrip('\n') + '\n```'
if isinstance(opt[attr], dict) and "_type" in opt[attr]:
if opt[attr]["_type"] == "literalExpression":
if "\n" in opt[attr]["text"]:
res = "\n```nix\n" + opt[attr]["text"].rstrip("\n") + "\n```"
else:
res = '```{}```'.format(opt[attr]['text'])
elif opt[attr]['_type'] == 'literalMD':
res = opt[attr]['text']
res = "```{}```".format(opt[attr]["text"])
elif opt[attr]["_type"] == "literalMD":
res = opt[attr]["text"]
else:
assert RuntimeError(f"Unhandled option type {opt[attr]["_type"]}")
else:
s = str(opt[attr])
if s == "":
res = '`""`'
elif '\n' in s:
res = '\n```\n' + s.rstrip('\n') + '\n```'
elif "\n" in s:
res = "\n```\n" + s.rstrip("\n") + "\n```"
else:
res = '```{}```'.format(s)
res = "```{}```".format(s)
return "- " + attr + ": " + res # type: ignore
return '- ' + attr + ': ' + res # type: ignore
def print_option(opt):
if isinstance(opt['description'], dict) and '_type' in opt['description']: # mdDoc
description = opt['description']['text']
if isinstance(opt["description"], dict) and "_type" in opt["description"]: # mdDoc
description = opt["description"]["text"]
else:
description = opt['description']
print(template.format(
key=opt['name'],
description=description or "",
type="- type: ```{}```".format(opt['type']),
default=render_option_value(opt, 'default'),
example=render_option_value(opt, 'example')))
description = opt["description"]
print(
template.format(
key=opt["name"],
description=description or "",
type="- type: ```{}```".format(opt["type"]),
default=render_option_value(opt, "default"),
example=render_option_value(opt, "example"),
)
)
print(header)
for opt in options:
if any([opt['name'].startswith(c) for c in groups]):
if any([opt["name"].startswith(c) for c in groups]):
continue
print_option(opt)
for c in groups:
print('## `{}`'.format(c))
print("## `{}`".format(c))
print()
for opt in options:
if opt['name'].startswith(c):
if opt["name"].startswith(c):
print_option(opt)

View File

@ -1,33 +1,37 @@
import smtplib, sys
import argparse
import os
import uuid
import imaplib
from datetime import datetime, timedelta
import email
import email.utils
import imaplib
import smtplib
import time
import uuid
from datetime import datetime, timedelta
from typing import cast
RETRY = 100
def _send_mail(smtp_host, smtp_port, smtp_username, from_addr, from_pwd, to_addr, subject, starttls):
print("Sending mail with subject '{}'".format(subject))
message = "\n".join([
"From: {from_addr}",
"To: {to_addr}",
"Subject: {subject}",
"Message-ID: {random}@mail-check.py",
"Date: {date}",
"",
"This validates our mail server can send to Gmail :/"]).format(
from_addr=from_addr,
to_addr=to_addr,
subject=subject,
random=str(uuid.uuid4()),
date=email.utils.formatdate(),
)
def _send_mail(
smtp_host, smtp_port, smtp_username, from_addr, from_pwd, to_addr, subject, starttls
):
print("Sending mail with subject '{}'".format(subject))
message = "\n".join(
[
"From: {from_addr}",
"To: {to_addr}",
"Subject: {subject}",
"Message-ID: {random}@mail-check.py",
"Date: {date}",
"",
"This validates our mail server can send to Gmail :/",
]
).format(
from_addr=from_addr,
to_addr=to_addr,
subject=subject,
random=str(uuid.uuid4()),
date=email.utils.formatdate(),
)
retry = RETRY
while True:
@ -44,7 +48,9 @@ def _send_mail(smtp_host, smtp_port, smtp_username, from_addr, from_pwd, to_addr
except smtplib.SMTPResponseException as e:
if e.smtp_code == 451: # service unavailable error
print(e)
elif e.smtp_code == 454: # smtplib.SMTPResponseException: (454, b'4.3.0 Try again later')
elif (
e.smtp_code == 454
): # smtplib.SMTPResponseException: (454, b'4.3.0 Try again later')
print(e)
else:
raise
@ -62,15 +68,17 @@ def _send_mail(smtp_host, smtp_port, smtp_username, from_addr, from_pwd, to_addr
print("Retry attempts exhausted")
exit(5)
def _read_mail(
imap_host,
imap_port,
imap_username,
to_pwd,
subject,
ignore_dkim_spf,
show_body=False,
delete=True):
imap_host,
imap_port,
imap_username,
to_pwd,
subject,
ignore_dkim_spf,
show_body=False,
delete=True,
):
print("Reading mail from %s" % imap_username)
message = None
@ -81,28 +89,31 @@ def _read_mail(
today = datetime.today()
cutoff = today - timedelta(days=1)
dt = cutoff.strftime('%d-%b-%Y')
dt = cutoff.strftime("%d-%b-%Y")
for _ in range(0, RETRY):
print("Retrying")
obj.select()
_, data = obj.search(None, '(SINCE %s) (SUBJECT "%s")' % (dt, subject))
if data == [b'']:
if data == [b""]:
time.sleep(1)
continue
uids = data[0].decode("utf-8").split(" ")
if len(uids) != 1:
print("Warning: %d messages have been found with subject containing %s " % (len(uids), subject))
print(
"Warning: %d messages have been found with subject containing %s "
% (len(uids), subject)
)
# FIXME: we only consider the first matching message...
uid = uids[0]
_, raw = obj.fetch(uid, '(RFC822)')
_, raw = obj.fetch(uid, "(RFC822)")
if delete:
obj.store(uid, '+FLAGS', '\\Deleted')
obj.store(uid, "+FLAGS", "\\Deleted")
obj.expunge()
assert raw[0] and raw[0][1]
message = email.message_from_bytes(cast(bytes, raw[0][1]))
print("Message with subject '%s' has been found" % message['subject'])
print("Message with subject '%s' has been found" % message["subject"])
if show_body:
if message.is_multipart():
for part in message.walk():
@ -118,21 +129,24 @@ def _read_mail(
break
if message is None:
print("Error: no message with subject '%s' has been found in INBOX of %s" % (subject, imap_username))
print(
"Error: no message with subject '%s' has been found in INBOX of %s"
% (subject, imap_username)
)
exit(1)
if ignore_dkim_spf:
return
# gmail set this standardized header
if 'ARC-Authentication-Results' in message:
if "dkim=pass" in message['ARC-Authentication-Results']:
if "ARC-Authentication-Results" in message:
if "dkim=pass" in message["ARC-Authentication-Results"]:
print("DKIM ok")
else:
print("Error: no DKIM validation found in message:")
print(message.as_string())
exit(2)
if "spf=pass" in message['ARC-Authentication-Results']:
if "spf=pass" in message["ARC-Authentication-Results"]:
print("SPF ok")
else:
print("Error: no SPF validation found in message:")
@ -142,71 +156,108 @@ def _read_mail(
print("DKIM and SPF verification failed")
exit(4)
def send_and_read(args):
src_pwd = None
if args.src_password_file is not None:
src_pwd = args.src_password_file.readline().rstrip()
dst_pwd = args.dst_password_file.readline().rstrip()
if args.imap_username != '':
if args.imap_username != "":
imap_username = args.imap_username
else:
imap_username = args.to_addr
subject = "{}".format(uuid.uuid4())
_send_mail(smtp_host=args.smtp_host,
smtp_port=args.smtp_port,
smtp_username=args.smtp_username,
from_addr=args.from_addr,
from_pwd=src_pwd,
to_addr=args.to_addr,
subject=subject,
starttls=args.smtp_starttls)
_send_mail(
smtp_host=args.smtp_host,
smtp_port=args.smtp_port,
smtp_username=args.smtp_username,
from_addr=args.from_addr,
from_pwd=src_pwd,
to_addr=args.to_addr,
subject=subject,
starttls=args.smtp_starttls,
)
_read_mail(
imap_host=args.imap_host,
imap_port=args.imap_port,
imap_username=imap_username,
to_pwd=dst_pwd,
subject=subject,
ignore_dkim_spf=args.ignore_dkim_spf,
)
_read_mail(imap_host=args.imap_host,
imap_port=args.imap_port,
imap_username=imap_username,
to_pwd=dst_pwd,
subject=subject,
ignore_dkim_spf=args.ignore_dkim_spf)
def read(args):
_read_mail(imap_host=args.imap_host,
imap_port=args.imap_port,
imap_username=args.imap_username,
to_pwd=args.imap_password,
subject=args.subject,
ignore_dkim_spf=args.ignore_dkim_spf,
show_body=args.show_body,
delete=False)
_read_mail(
imap_host=args.imap_host,
imap_port=args.imap_port,
imap_username=args.imap_username,
to_pwd=args.imap_password,
subject=args.subject,
ignore_dkim_spf=args.ignore_dkim_spf,
show_body=args.show_body,
delete=False,
)
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers()
parser_send_and_read = subparsers.add_parser('send-and-read', description="Send a email with a subject containing a random UUID and then try to read this email from the recipient INBOX.")
parser_send_and_read.add_argument('--smtp-host', type=str)
parser_send_and_read.add_argument('--smtp-port', type=str, default=25)
parser_send_and_read.add_argument('--smtp-starttls', action='store_true')
parser_send_and_read.add_argument('--smtp-username', type=str, default='', help="username used for smtp login. If not specified, the from-addr value is used")
parser_send_and_read.add_argument('--from-addr', type=str)
parser_send_and_read.add_argument('--imap-host', required=True, type=str)
parser_send_and_read.add_argument('--imap-port', type=str, default=993)
parser_send_and_read.add_argument('--to-addr', type=str, required=True)
parser_send_and_read.add_argument('--imap-username', type=str, default='', help="username used for imap login. If not specified, the to-addr value is used")
parser_send_and_read.add_argument('--src-password-file', type=argparse.FileType('r'))
parser_send_and_read.add_argument('--dst-password-file', required=True, type=argparse.FileType('r'))
parser_send_and_read.add_argument('--ignore-dkim-spf', action='store_true', help="to ignore the dkim and spf verification on the read mail")
parser_send_and_read = subparsers.add_parser(
"send-and-read",
description="Send a email with a subject containing a random UUID and then try to read this email from the recipient INBOX.",
)
parser_send_and_read.add_argument("--smtp-host", type=str)
parser_send_and_read.add_argument("--smtp-port", type=str, default=25)
parser_send_and_read.add_argument("--smtp-starttls", action="store_true")
parser_send_and_read.add_argument(
"--smtp-username",
type=str,
default="",
help="username used for smtp login. If not specified, the from-addr value is used",
)
parser_send_and_read.add_argument("--from-addr", type=str)
parser_send_and_read.add_argument("--imap-host", required=True, type=str)
parser_send_and_read.add_argument("--imap-port", type=str, default=993)
parser_send_and_read.add_argument("--to-addr", type=str, required=True)
parser_send_and_read.add_argument(
"--imap-username",
type=str,
default="",
help="username used for imap login. If not specified, the to-addr value is used",
)
parser_send_and_read.add_argument("--src-password-file", type=argparse.FileType("r"))
parser_send_and_read.add_argument(
"--dst-password-file", required=True, type=argparse.FileType("r")
)
parser_send_and_read.add_argument(
"--ignore-dkim-spf",
action="store_true",
help="to ignore the dkim and spf verification on the read mail",
)
parser_send_and_read.set_defaults(func=send_and_read)
parser_read = subparsers.add_parser('read', description="Search for an email with a subject containing 'subject' in the INBOX.")
parser_read.add_argument('--imap-host', type=str, default="localhost")
parser_read.add_argument('--imap-port', type=str, default=993)
parser_read.add_argument('--imap-username', required=True, type=str)
parser_read.add_argument('--imap-password', required=True, type=str)
parser_read.add_argument('--ignore-dkim-spf', action='store_true', help="to ignore the dkim and spf verification on the read mail")
parser_read.add_argument('--show-body', action='store_true', help="print mail text/plain payload")
parser_read.add_argument('subject', type=str)
parser_read = subparsers.add_parser(
"read",
description="Search for an email with a subject containing 'subject' in the INBOX.",
)
parser_read.add_argument("--imap-host", type=str, default="localhost")
parser_read.add_argument("--imap-port", type=str, default=993)
parser_read.add_argument("--imap-username", required=True, type=str)
parser_read.add_argument("--imap-password", required=True, type=str)
parser_read.add_argument(
"--ignore-dkim-spf",
action="store_true",
help="to ignore the dkim and spf verification on the read mail",
)
parser_read.add_argument(
"--show-body", action="store_true", help="print mail text/plain payload"
)
parser_read.add_argument("subject", type=str)
parser_read.set_defaults(func=read)
args = parser.parse_args()