LibertaCasa/system
9
0
Fork 0
Code Issues 3 Pull Requests Projects Releases Wiki Activity

mxme.py: Checking/Enabling secondary MX.

Browse Source 
Signed-off-by: Georg <georg@lysergic.dev>
This commit is contained in:
Georg Pfuetzenreuter 2021-09-15 14:18:48 +02:00
parent 8a9fe75e61
commit 89524ab066
Signed by: Georg
GPG Key ID: 1DAF57F49F8E8F22
1 changed files with 177 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

220
scripts/python/mxme.py
View File

@ -1,5 +1,20 @@
#!/usr/bin/python3 #!/usr/bin/python3
""" """
The flagship email-domain enabler and repair tool for Mailcow/PowerDNS infrastructures.
The following checks will be performed:
DNS:
- DMARC record
- DKIM record
- MX1 + MX2 records
Mail:
- MX1 domain entry
- MX2 domain entry
- MX1 DKIM keypair
If any of the checks fail, the failed operation will be written to the respective system, in addition to overwriting possibly existing DNS records with fresh DMARC/DKIM/SPF/MX records.
Created and Last modified: 15/09/2021 by Georg Pfuetzenreuter <georg@lysergic.dev>
""" """
import requests import requests
import sys import sys
@ -21,8 +36,10 @@ APIKEY_PDNS = os.environ.get('APIKEY_PDNS')
# MAILCOW SETTINGS # MAILCOW SETTINGS
ENDPOINT_MAILCOW = os.environ.get('ENDPOINT_MAILCOW') ENDPOINT_MAILCOW = os.environ.get('ENDPOINT_MAILCOW')
APIKEY_MAILCOW = os.environ.get('APIKEY_MAILCOW') APIKEY_MAILCOW = os.environ.get('APIKEY_MAILCOW')
ENDPOINT_MAILCOW_2 = os.environ.get('ENDPOINT_MAILCOW_2')
APIKEY_MAILCOW_2 = os.environ.get('APIKEY_MAILCOW_2')
if None in (ENDPOINT_PDNS, APIKEY_PDNS, ENDPOINT_MAILCOW, APIKEY_MAILCOW): if None in (ENDPOINT_PDNS, APIKEY_PDNS, ENDPOINT_MAILCOW, APIKEY_MAILCOW, ENDPOINT_MAILCOW_2, APIKEY_MAILCOW_2):
print("Could not load environment variables. Please check your .env file.") print("Could not load environment variables. Please check your .env file.")
sys.exit(0) sys.exit(0)
@ -58,7 +75,7 @@ if dnsok == False:
sys.exit(1) sys.exit(1)
# MAILCOW (can I put cow emoji in comment?) # MAILCOW (can I put cow emoji in comment?)
print("Mail: Querying domain ...") print("Mail [MX1]: Querying domain ...")
server = ENDPOINT_MAILCOW server = ENDPOINT_MAILCOW
api_key = APIKEY_MAILCOW api_key = APIKEY_MAILCOW
api = '/api/v1' api = '/api/v1'
@ -91,36 +108,67 @@ except requests.exceptions.HTTPError as err:
print(err) print(err)
sys.exit(1) sys.exit(1)
if mailok == True and initprimary == False: print("Mail [MX2]: Querying domain ...")
print("Mail: Querying DKIM ...") server2 = ENDPOINT_MAILCOW_2
URL = server + get + '/dkim/' + domain api_key2 = APIKEY_MAILCOW_2
try: api2 = '/api/v1'
response = requests.get( URL = server2 + get + '/domain/' + domain
URL, try:
headers = {'accept': 'application/json', 'X-API-Key': api_key}, response = requests.get(
) URL,
data = response.json() headers = {'accept': 'application/json', 'X-API-Key': api_key2},
#print(data) )
if 'dkim_selector' in data: data = response.json()
selector = data['dkim_selector'] status = response.status_code
txtshould = data['dkim_txt'] #print(data)
length = data['length'] if 'max_new_mailbox_quota' in data:
pubkey = data['pubkey'] print("Mail: Domain found.")
#print(f"Domain: {domain}\nSelector: {selector}\nTXT: {txtshould}\nPublic Key: {pubkey}") mail2ok = True
#print(txtshould) initsecondary = False
print("Mail: DKIM keypair found.") domain_name = data['domain_name']
dkimok = True relayhost = data['relayhost']
else: else:
dkimok = False mail2ok = False
print("Mail: No DKIM keypair found.") initsecondary = True
except KeyError: print("Mail: Domain NOT found.")
print("Mail: No or faulty DKIM lookup.") dkim2ok = False
except requests.exceptions.ConnectionError as err: except requests.exceptions.ConnectionError as err:
print("Connection failed.") print("Connection failed.")
sys.exit(1) sys.exit(1)
except requests.exceptions.HTTPError as err: except requests.exceptions.HTTPError as err:
print(err) print(err)
sys.exit(1) sys.exit(1)
#if mailok == True and initprimary == False:
print("Mail: Querying DKIM ...")
URL = server + get + '/dkim/' + domain
try:
response = requests.get(
URL,
headers = {'accept': 'application/json', 'X-API-Key': api_key},
)
data = response.json()
#print(data)
if 'dkim_selector' in data:
selector = data['dkim_selector']
txtshould = data['dkim_txt']
length = data['length']
pubkey = data['pubkey']
#print(f"Domain: {domain}\nSelector: {selector}\nTXT: {txtshould}\nPublic Key: {pubkey}")
#print(txtshould)
print("Mail: DKIM keypair found.")
dkimok = True
else:
dkimok = False
print("Mail: No DKIM keypair found.")
except KeyError:
print("Mail: No or faulty DKIM lookup.")
except requests.exceptions.ConnectionError as err:
print("Connection failed.")
sys.exit(1)
except requests.exceptions.HTTPError as err:
print(err)
sys.exit(1)
print("DNS: Querying records ...") print("DNS: Querying records ...")
URL = ENDPOINT_PDNS + '/api/v1/servers/localhost/zones/' + domain + './export' URL = ENDPOINT_PDNS + '/api/v1/servers/localhost/zones/' + domain + './export'
@ -155,10 +203,32 @@ try:
except NameError: except NameError:
print("DNS: No DMARC TXT record found.") print("DNS: No DMARC TXT record found.")
dnsdmarcok = False dnsdmarcok = False
for record in data.split('\n'):
if '10 3gy.de.' in record:
mxrec1 = record
try:
mxrec1
print("DNS: Found primary MX record.")
dnsmx1ok = True
except NameError:
print("DNS: Did NOT find primary MX record.")
dnsmx1ok = False
for record in data.split('\n'):
if '20 3gy.pl.' in record:
mxrec2 = record
try:
mxrec2
print("DNS: Found secondary MX record.")
dnsmx2ok = True
except NameError:
print("DNS: Did NOT find secondary MX record.")
dnsmx2ok = False
except NameError: except NameError:
print("DNS: Missing or faulty DKIM/DMARC records.") print("DNS: Missing or faulty records.")
dnsdmarcok = False dnsdmarcok = False
dnsdkimok = False dnsdkimok = False
dnsmx1ok = False
dnsmx2ok = False
except requests.exceptions.ConnectionError as err: except requests.exceptions.ConnectionError as err:
print("Connection failed.") print("Connection failed.")
sys.exit(1) sys.exit(1)
@ -166,17 +236,21 @@ except requests.exceptions.HTTPError as err:
print(err) print(err)
sys.exit(1) sys.exit(1)
if dnsok == True and mailok == True and dkimok == True and dnsdkimok == True and dnsdmarcok == True: if dnsok == True and mailok == True and dkimok == True and dnsdkimok == True and dnsdmarcok == True and dnsmx1ok == True and dnsmx2ok == True:
print("All good. No changes seem to be needed. Aborting.") print("All good. No changes seem to be needed. Aborting.")
sys.exit(0) sys.exit(0)
else: else:
print("Found inconsistencies:") print("Found inconsistencies:")
print(f"DNS OK: {dnsok} - Mail OK: {mailok} - Mail DKIM OK: {dkimok} - DNS DKIM OK: {dnsdkimok} - DNS DMARC OK: {dnsdmarcok}") print(f"DNS OK: {dnsok} - Mail OK: {mailok} - Mail DKIM OK: {dkimok} - DNS DKIM OK: {dnsdkimok} - DNS DMARC OK: {dnsdmarcok} - DNS MX1 OK: {dnsmx1ok} - DNS MX2 OK: {dnsmx2ok}")
print("Will attempt a full repair if not cancelled within 5 seconds ...") try:
time.sleep(5) print("Will attempt a full repair if not cancelled within 5 seconds ...")
time.sleep(5)
except KeyboardInterrupt:
print(" Cancelled!")
sys.exit(1)
if initprimary == True: if initprimary == True:
print("Mail: Initializing domain ...") print("Mail [MX1]: Initializing domain ...")
URL = server + add + '/domain' URL = server + add + '/domain'
payload = { payload = {
"active": "1", "active": "1",
@ -199,21 +273,60 @@ if initprimary == True:
json = payload, json = payload,
) )
data = response.json() data = response.json()
status = data[0]['type']
try: try:
status = data[0]['type']
status status
except: except:
print("Mail Error:") print("Mail [MX1] Error:")
print(data) print(data)
sys.exit(1) sys.exit(1)
if status == 'success': if status == 'success':
print("Mail: Created domain.") print("Mail [MX1]: Created domain.")
if status == 'danger': if status == 'danger':
print("Mail: Failed to create domain.") print("Mail [MX1]: Failed to create domain.")
print(data) print(data)
#print(f"CREATION: {status}") #print(f"CREATION: {status}")
if initprimary == True or dkimok == False: if initsecondary == True:
print("Mail [MX2]: Initializing domain ...")
URL = server2 + add + '/domain'
payload = {
"active": "1",
"aliases": "20",
"backupmx": "1",
"defquota": "1024",
"description": domain + ' - Failover',
"domain": domain,
"mailboxes": "10",
"maxquota": "2048",
"quota": "5120",
"relay_all_recipients": "1",
"rl_frame": "s",
"rl_value": "10",
"restart_sogo": "0"
}
response = requests.post(
URL,
headers = {'accept': 'application/json', 'X-API-Key': api_key2, 'Content-Type': 'application/json'},
json = payload,
)
data = response.json()
try:
status = data[0]['type']
status
except:
print("Mail [MX2] Error:")
print(data)
sys.exit(1)
if status == 'success':
print("Mail [MX2]: Created domain.")
if status == 'danger':
print("Mail [MX2]: Failed to create domain.")
print(data)
#print(f"CREATION: {status}")
#if initprimary == True or dkimok == False:
if dkimok == False:
print("Mail: Initializing DKIM ...") print("Mail: Initializing DKIM ...")
URL = server + add + '/dkim' URL = server + add + '/dkim'
payload = { payload = {
@ -340,6 +453,27 @@ else:
print(status) print(status)
print(response.json()) print(response.json())
sys.exit(1) sys.exit(1)
print("DNS: Patching MX ...")
payload = {
"rrsets": [{"name": domain + ".", "type": "MX", "ttl": "3600", "changetype": "REPLACE", "records": [{"content": "10 3gy.de.", "disabled": False, "name": domain + "."}, {"content": "20 3gy.pl.", "disabled": False, "name": domain + "."}]}]
}
response = requests.patch(
URL,
headers = {'accept': 'application/json', 'X-API-Key': APIKEY_PDNS, 'Content-Type': 'application/json'},
json = payload,
)
status = response.status_code
if status == 204:
print("MX: OK!")
elif status == 422:
print("MX: Failed:")
print(response.json())
sys.exit(1)
else:
print("Unhandled error.")
print(status)
print(response.json())
sys.exit(1)
print("Done.") print("Done.")
sys.exit(0) sys.exit(0)