mirror of
https://github.com/mail-in-a-box/mailinabox.git
synced 2026-09-12 00:18:59 +02:00
First big pass on PEP8'ing all the things.
All PEP8 errors (except line length) have been fixed except one. That one will require a little bit of refactoring.
This commit is contained in:
@@ -7,100 +7,110 @@
|
||||
# where ipaddr is the IP address of your Mail-in-a-Box
|
||||
# and hostname is the domain name to check the DNS for.
|
||||
|
||||
import sys, re, difflib
|
||||
import dns.reversename, dns.resolver
|
||||
import sys
|
||||
import re
|
||||
import difflib
|
||||
import dns.reversename
|
||||
import dns.resolver
|
||||
|
||||
if len(sys.argv) < 3:
|
||||
print("Usage: tests/dns.py ipaddress hostname [primary hostname]")
|
||||
sys.exit(1)
|
||||
print("Usage: tests/dns.py ipaddress hostname [primary hostname]")
|
||||
sys.exit(1)
|
||||
|
||||
ipaddr, hostname = sys.argv[1:3]
|
||||
primary_hostname = hostname
|
||||
if len(sys.argv) == 4:
|
||||
primary_hostname = sys.argv[3]
|
||||
primary_hostname = sys.argv[3]
|
||||
|
||||
|
||||
def test(server, description):
|
||||
tests = [
|
||||
(hostname, "A", ipaddr),
|
||||
#(hostname, "NS", "ns1.%s.;ns2.%s." % (primary_hostname, primary_hostname)),
|
||||
("ns1." + primary_hostname, "A", ipaddr),
|
||||
("ns2." + primary_hostname, "A", ipaddr),
|
||||
("www." + hostname, "A", ipaddr),
|
||||
(hostname, "MX", "10 " + primary_hostname + "."),
|
||||
(hostname, "TXT", "\"v=spf1 mx -all\""),
|
||||
("mail._domainkey." + hostname, "TXT", "\"v=DKIM1; k=rsa; s=email; \" \"p=__KEY__\""),
|
||||
#("_adsp._domainkey." + hostname, "TXT", "\"dkim=all\""),
|
||||
("_dmarc." + hostname, "TXT", "\"v=DMARC1; p=quarantine\""),
|
||||
]
|
||||
return test2(tests, server, description)
|
||||
tests = [
|
||||
(hostname, "A", ipaddr),
|
||||
#(hostname, "NS", "ns1.%s.;ns2.%s." % (primary_hostname, primary_hostname)),
|
||||
("ns1." + primary_hostname, "A", ipaddr),
|
||||
("ns2." + primary_hostname, "A", ipaddr),
|
||||
("www." + hostname, "A", ipaddr),
|
||||
(hostname, "MX", "10 " + primary_hostname + "."),
|
||||
(hostname, "TXT", "\"v=spf1 mx -all\""),
|
||||
("mail._domainkey." + hostname, "TXT", "\"v=DKIM1; k=rsa; s=email; \" \"p=__KEY__\""),
|
||||
#("_adsp._domainkey." + hostname, "TXT", "\"dkim=all\""),
|
||||
("_dmarc." + hostname, "TXT", "\"v=DMARC1; p=quarantine\""),
|
||||
]
|
||||
return test2(tests, server, description)
|
||||
|
||||
|
||||
def test_ptr(server, description):
|
||||
ipaddr_rev = dns.reversename.from_address(ipaddr)
|
||||
tests = [
|
||||
(ipaddr_rev, "PTR", hostname+'.'),
|
||||
]
|
||||
return test2(tests, server, description)
|
||||
ipaddr_rev = dns.reversename.from_address(ipaddr)
|
||||
tests = [
|
||||
(ipaddr_rev, "PTR", hostname+'.'),
|
||||
]
|
||||
return test2(tests, server, description)
|
||||
|
||||
|
||||
def test2(tests, server, description):
|
||||
first = True
|
||||
resolver = dns.resolver.get_default_resolver()
|
||||
resolver.nameservers = [server]
|
||||
for qname, rtype, expected_answer in tests:
|
||||
# do the query and format the result as a string
|
||||
try:
|
||||
response = dns.resolver.query(qname, rtype)
|
||||
except dns.resolver.NoNameservers:
|
||||
# host did not have an answer for this query
|
||||
print("Could not connect to %s for DNS query." % server)
|
||||
sys.exit(1)
|
||||
except (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
|
||||
response = ["[no value]"]
|
||||
response = ";".join(str(r) for r in response)
|
||||
response = re.sub(r"(\"p=).*(\")", r"\1__KEY__\2", response) # normalize DKIM key
|
||||
response = response.replace("\"\" ", "") # normalize TXT records (DNSSEC signing inserts empty text string components)
|
||||
first = True
|
||||
resolver = dns.resolver.get_default_resolver()
|
||||
resolver.nameservers = [server]
|
||||
for qname, rtype, expected_answer in tests:
|
||||
# do the query and format the result as a string
|
||||
try:
|
||||
response = dns.resolver.query(qname, rtype)
|
||||
except dns.resolver.NoNameservers:
|
||||
# host did not have an answer for this query
|
||||
print("Could not connect to %s for DNS query." % server)
|
||||
sys.exit(1)
|
||||
except (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
|
||||
response = ["[no value]"]
|
||||
response = ";".join(str(r) for r in response)
|
||||
# normalize DKIM key
|
||||
response = re.sub(r"(\"p=).*(\")", r"\1__KEY__\2", response)
|
||||
# normalize TXT records (DNSSEC signing inserts empty text
|
||||
# string components)
|
||||
response = response.replace("\"\" ", "")
|
||||
|
||||
# is it right?
|
||||
if response == expected_answer:
|
||||
#print(server, ":", qname, rtype, "?", response)
|
||||
continue
|
||||
# is it right?
|
||||
if response == expected_answer:
|
||||
#print(server, ":", qname, rtype, "?", response)
|
||||
continue
|
||||
|
||||
# show prolem
|
||||
if first:
|
||||
print("Incorrect DNS Response from", description)
|
||||
print()
|
||||
print("QUERY ", "RESPONSE ", "CORRECT VALUE", sep='\t')
|
||||
first = False
|
||||
# show problem
|
||||
if first:
|
||||
print("Incorrect DNS Response from", description)
|
||||
print()
|
||||
print("QUERY ", "RESPONSE ", "CORRECT VALUE", sep='\t')
|
||||
first = False
|
||||
|
||||
print((qname + "/" + rtype).ljust(20), response.ljust(12), expected_answer, sep='\t')
|
||||
return first # success
|
||||
print((qname + "/" + rtype).ljust(20), response.ljust(12), expected_answer, sep='\t')
|
||||
# success
|
||||
return first
|
||||
|
||||
# Test the response from the machine itself.
|
||||
if not test(ipaddr, "Mail-in-a-Box"):
|
||||
print ()
|
||||
print ("Please run the Mail-in-a-Box setup script on %s again." % hostname)
|
||||
sys.exit(1)
|
||||
print ()
|
||||
print ("Please run the Mail-in-a-Box setup script on %s again." % hostname)
|
||||
sys.exit(1)
|
||||
else:
|
||||
print ("The Mail-in-a-Box provided correct DNS answers.")
|
||||
print ()
|
||||
print ("The Mail-in-a-Box provided correct DNS answers.")
|
||||
print ()
|
||||
|
||||
# If those settings are OK, also test Google's Public DNS
|
||||
# to see if the machine is hooked up to recursive DNS properly.
|
||||
if not test("8.8.8.8", "Google Public DNS"):
|
||||
print ()
|
||||
print ("Check that the nameserver settings for %s are correct at your domain registrar. It may take a few hours for Google Public DNS to update after changes on your Mail-in-a-Box." % hostname)
|
||||
sys.exit(1)
|
||||
else:
|
||||
print ("Your domain registrar or DNS host appears to be configured correctly as well. Public DNS provides the same answers.")
|
||||
print ()
|
||||
# If those settings are OK, also test Google's Public DNS
|
||||
# to see if the machine is hooked up to recursive DNS properly.
|
||||
if not test("8.8.8.8", "Google Public DNS"):
|
||||
print ()
|
||||
print ("Check that the nameserver settings for %s are correct at your domain registrar. It may take a few hours for Google Public DNS to update after changes on your Mail-in-a-Box." % hostname)
|
||||
sys.exit(1)
|
||||
else:
|
||||
print ("Your domain registrar or DNS host appears to be configured correctly as well. Public DNS provides the same answers.")
|
||||
print ()
|
||||
|
||||
# And if that's OK, also check reverse DNS (the PTR record).
|
||||
if not test_ptr("8.8.8.8", "Google Public DNS (Reverse DNS)"):
|
||||
print ()
|
||||
print ("The reverse DNS for %s is not correct. Consult your ISP for how to set the reverse DNS (also called the PTR record) for %s to %s." % (hostname, hostname, ipaddr))
|
||||
sys.exit(1)
|
||||
else:
|
||||
print ("And the reverse DNS for the domain is correct.")
|
||||
print ()
|
||||
print ("DNS is OK.")
|
||||
# And if that's OK, also check reverse DNS (the PTR record).
|
||||
if not test_ptr("8.8.8.8", "Google Public DNS (Reverse DNS)"):
|
||||
print ()
|
||||
print ("The reverse DNS for %s is not correct. Consult your ISP for how to set the reverse DNS (also called the PTR record) for %s to %s." % (hostname, hostname, ipaddr))
|
||||
sys.exit(1)
|
||||
else:
|
||||
print ("And the reverse DNS for the domain is correct.")
|
||||
print ()
|
||||
print ("DNS is OK.")
|
||||
|
||||
@@ -1,28 +1,34 @@
|
||||
#!/usr/bin/env python3
|
||||
# Tests sending and receiving mail by sending a test message to yourself.
|
||||
|
||||
import sys, imaplib, smtplib, uuid, time
|
||||
import socket, dns.reversename, dns.resolver
|
||||
import sys
|
||||
import imaplib
|
||||
import smtplib
|
||||
import uuid
|
||||
import time
|
||||
import socket
|
||||
import dns.reversename
|
||||
import dns.resolver
|
||||
|
||||
if len(sys.argv) < 3:
|
||||
print("Usage: tests/mail.py hostname emailaddress password")
|
||||
sys.exit(1)
|
||||
print("Usage: tests/mail.py hostname emailaddress password")
|
||||
sys.exit(1)
|
||||
|
||||
host, emailaddress, pw = sys.argv[1:4]
|
||||
|
||||
# Attempt to login with IMAP. Our setup uses email addresses
|
||||
# as IMAP/SMTP usernames.
|
||||
try:
|
||||
M = imaplib.IMAP4_SSL(host)
|
||||
M.login(emailaddress, pw)
|
||||
M = imaplib.IMAP4_SSL(host)
|
||||
M.login(emailaddress, pw)
|
||||
except OSError as e:
|
||||
print("Connection error:", e)
|
||||
sys.exit(1)
|
||||
print("Connection error:", e)
|
||||
sys.exit(1)
|
||||
except imaplib.IMAP4.error as e:
|
||||
# any sort of login error
|
||||
e = ", ".join(a.decode("utf8") for a in e.args)
|
||||
print("IMAP error:", e)
|
||||
sys.exit(1)
|
||||
# any sort of login error
|
||||
e = ", ".join(a.decode("utf8") for a in e.args)
|
||||
print("IMAP error:", e)
|
||||
sys.exit(1)
|
||||
|
||||
M.select()
|
||||
print("IMAP login is OK.")
|
||||
@@ -35,10 +41,10 @@ To: {emailto}
|
||||
Subject: {subject}
|
||||
|
||||
This is a test message. It should be automatically deleted by the test script.""".format(
|
||||
emailaddress=emailaddress,
|
||||
emailto=emailto,
|
||||
subject=mailsubject,
|
||||
)
|
||||
emailaddress=emailaddress,
|
||||
emailto=emailto,
|
||||
subject=mailsubject,
|
||||
)
|
||||
|
||||
# Connect to the server on the SMTP submission TLS port.
|
||||
server = smtplib.SMTP(host, 587)
|
||||
@@ -46,20 +52,21 @@ server = smtplib.SMTP(host, 587)
|
||||
server.starttls()
|
||||
|
||||
# Verify that the EHLO name matches the server's reverse DNS.
|
||||
ipaddr = socket.gethostbyname(host) # IPv4 only!
|
||||
reverse_ip = dns.reversename.from_address(ipaddr) # e.g. "1.0.0.127.in-addr.arpa."
|
||||
ipaddr = socket.gethostbyname(host) # IPv4 only!
|
||||
reverse_ip = dns.reversename.from_address(ipaddr) # e.g. "1.0.0.127.in-addr.arpa."
|
||||
|
||||
try:
|
||||
reverse_dns = dns.resolver.query(reverse_ip, 'PTR')[0].target.to_text(omit_final_dot=True) # => hostname
|
||||
reverse_dns = dns.resolver.query(reverse_ip, 'PTR')[0].target.to_text(omit_final_dot=True) # => hostname
|
||||
except dns.resolver.NXDOMAIN:
|
||||
print("Reverse DNS lookup failed for %s. SMTP EHLO name check skipped." % ipaddr)
|
||||
reverse_dns = None
|
||||
print("Reverse DNS lookup failed for %s. SMTP EHLO name check skipped." % ipaddr)
|
||||
reverse_dns = None
|
||||
if reverse_dns is not None:
|
||||
server.ehlo_or_helo_if_needed() # must send EHLO before getting the server's EHLO name
|
||||
helo_name = server.ehlo_resp.decode("utf8").split("\n")[0] # first line is the EHLO name
|
||||
if helo_name != reverse_dns:
|
||||
print("The server's EHLO name does not match its reverse hostname. Check DNS settings.")
|
||||
else:
|
||||
print("SMTP EHLO name (%s) is OK." % helo_name)
|
||||
server.ehlo_or_helo_if_needed() # must send EHLO before getting the server's EHLO name
|
||||
helo_name = server.ehlo_resp.decode("utf8").split("\n")[0] # first line is the EHLO name
|
||||
if helo_name != reverse_dns:
|
||||
print("The server's EHLO name does not match its reverse hostname. Check DNS settings.")
|
||||
else:
|
||||
print("SMTP EHLO name (%s) is OK." % helo_name)
|
||||
|
||||
# Login and send a test email.
|
||||
server.login(emailaddress, pw)
|
||||
@@ -68,40 +75,40 @@ server.quit()
|
||||
print("SMTP submission is OK.")
|
||||
|
||||
while True:
|
||||
# Wait so the message can propagate to the inbox.
|
||||
time.sleep(10)
|
||||
# Wait so the message can propagate to the inbox.
|
||||
time.sleep(10)
|
||||
|
||||
# Read the subject lines of all of the emails in the inbox
|
||||
# to find our test message, and then delete it.
|
||||
found = False
|
||||
typ, data = M.search(None, 'ALL')
|
||||
for num in data[0].split():
|
||||
typ, data = M.fetch(num, '(BODY[HEADER.FIELDS (SUBJECT)])')
|
||||
imapsubjectline = data[0][1].strip().decode("utf8")
|
||||
if imapsubjectline == "Subject: " + mailsubject:
|
||||
# We found our test message.
|
||||
found = True
|
||||
# Read the subject lines of all of the emails in the inbox
|
||||
# to find our test message, and then delete it.
|
||||
found = False
|
||||
typ, data = M.search(None, 'ALL')
|
||||
for num in data[0].split():
|
||||
typ, data = M.fetch(num, '(BODY[HEADER.FIELDS (SUBJECT)])')
|
||||
imapsubjectline = data[0][1].strip().decode("utf8")
|
||||
if imapsubjectline == "Subject: " + mailsubject:
|
||||
# We found our test message.
|
||||
found = True
|
||||
|
||||
# To test DKIM, download the whole mssage body. Unfortunately,
|
||||
# pydkim doesn't actually work.
|
||||
# You must 'sudo apt-get install python3-dkim python3-dnspython' first.
|
||||
#typ, msgdata = M.fetch(num, '(RFC822)')
|
||||
#msg = msgdata[0][1]
|
||||
#if dkim.verify(msg):
|
||||
# print("DKIM signature on the test message is OK (verified).")
|
||||
#else:
|
||||
# print("DKIM signature on the test message failed verification.")
|
||||
# To test DKIM, download the whole mssage body. Unfortunately,
|
||||
# pydkim doesn't actually work.
|
||||
# You must 'sudo apt-get install python3-dkim python3-dnspython' first.
|
||||
#typ, msgdata = M.fetch(num, '(RFC822)')
|
||||
#msg = msgdata[0][1]
|
||||
#if dkim.verify(msg):
|
||||
# print("DKIM signature on the test message is OK (verified).")
|
||||
#else:
|
||||
# print("DKIM signature on the test message failed verification.")
|
||||
|
||||
# Delete the test message.
|
||||
M.store(num, '+FLAGS', '\\Deleted')
|
||||
M.expunge()
|
||||
# Delete the test message.
|
||||
M.store(num, '+FLAGS', '\\Deleted')
|
||||
M.expunge()
|
||||
|
||||
break
|
||||
break
|
||||
|
||||
if found:
|
||||
break
|
||||
if found:
|
||||
break
|
||||
|
||||
print("Test message not present in the inbox yet...")
|
||||
print("Test message not present in the inbox yet...")
|
||||
|
||||
M.close()
|
||||
M.logout()
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
#!/usr/bin/env python3
|
||||
import smtplib, sys
|
||||
import smtplib
|
||||
import sys
|
||||
|
||||
if len(sys.argv) < 3:
|
||||
print("Usage: tests/smtp_server.py host email.to email.from")
|
||||
@@ -16,4 +17,3 @@ server = smtplib.SMTP(host, 25)
|
||||
server.set_debuglevel(1)
|
||||
server.sendmail(fromaddr, [toaddr], msg)
|
||||
server.quit()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user