summaryrefslogtreecommitdiffstats
path: root/scripts/python
diff options
context:
space:
mode:
Diffstat (limited to 'scripts/python')
-rwxr-xr-xscripts/python/mxme.py345
1 files changed, 345 insertions, 0 deletions
diff --git a/scripts/python/mxme.py b/scripts/python/mxme.py
new file mode 100755
index 0000000..6bdbf69
--- /dev/null
+++ b/scripts/python/mxme.py
@@ -0,0 +1,345 @@
+#!/usr/bin/python3
+"""
+"""
+import requests
+import sys
+import os
+from dotenv import load_dotenv
+import time
+
+if len(sys.argv) > 1:
+ domain = sys.argv[1]
+else:
+ print("Specify the domain name")
+ sys.exit(1)
+
+load_dotenv()
+# POWERDNS SETTINGS
+ENDPOINT_PDNS = os.environ.get('ENDPOINT_PDNS')
+APIKEY_PDNS = os.environ.get('APIKEY_PDNS')
+
+# MAILCOW SETTINGS
+ENDPOINT_MAILCOW = os.environ.get('ENDPOINT_MAILCOW')
+APIKEY_MAILCOW = os.environ.get('APIKEY_MAILCOW')
+
+if None in (ENDPOINT_PDNS, APIKEY_PDNS, ENDPOINT_MAILCOW, APIKEY_MAILCOW):
+ print("Could not load environment variables. Please check your .env file.")
+ sys.exit(0)
+
+print("Scanning " + domain)
+
+# QUERY POWERDNS
+print("DNS: Querying zone ...")
+URL = ENDPOINT_PDNS + '/api/v1/servers/localhost/zones/' + domain + './export'
+try:
+ response = requests.get(
+ URL,
+ headers = {'accept': 'text/plain', 'X-API-Key': APIKEY_PDNS},
+ )
+ response.raise_for_status()
+ data = response.text
+ #print(data)
+ status = response.status_code
+ if status == 200:
+ dnsok = True
+ print("DNS: Zone found.")
+ else:
+ dnsok = False
+ print("DNS: No zone found, or faulty lookup. Aborting.")
+ sys.exit(1)
+except requests.exceptions.HTTPError as err:
+ #print("No zone for this domain exists.")
+ raise SystemExit(err)
+ sys.exit(1)
+except requests.exceptions.ConnectionError as err:
+ print("Connection failed.")
+ sys.exit(1)
+if dnsok == False:
+ sys.exit(1)
+
+# MAILCOW (can I put cow emoji in comment?)
+print("Mail: Querying domain ...")
+server = ENDPOINT_MAILCOW
+api_key = APIKEY_MAILCOW
+api = '/api/v1'
+get = api + '/get'
+add = api + '/add'
+URL = server + get + '/domain/' + domain
+try:
+ response = requests.get(
+ URL,
+ headers = {'accept': 'application/json', 'X-API-Key': api_key},
+ )
+ data = response.json()
+ status = response.status_code
+ #print(data)
+ if 'max_new_mailbox_quota' in data:
+ print("Mail: Domain found.")
+ mailok = True
+ initprimary = False
+ domain_name = data['domain_name']
+ relayhost = data['relayhost']
+ else:
+ mailok = False
+ initprimary = True
+ print("Mail: Domain NOT found.")
+ dkimok = False
+except requests.exceptions.ConnectionError as err:
+ print("Connection failed.")
+ sys.exit(1)
+except requests.exceptions.HTTPError as err:
+ print(err)
+ 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 ...")
+URL = ENDPOINT_PDNS + '/api/v1/servers/localhost/zones/' + domain + './export'
+try:
+ response = requests.get(
+ URL,
+ headers = {'accept': 'text/plain', 'X-API-Key': APIKEY_PDNS},
+ )
+ data = response.text
+ #print(data)
+ for record in data.split('\n'):
+ txtsel = selector + '._'
+ if txtsel in record:
+ #print(record)
+ txtis = record.split('"')[1]
+ try:
+ txtis
+ print("DNS: Found DKIM TXT record.")
+ dnsdkimok = True
+ except NameError:
+ print("DNS: No DKIM TXT record found.")
+ dnsdkimok = False
+ for record in data.split('\n'):
+ txtsel = selector + '._'
+ if '_dmarc' in record:
+ #print(record)
+ txtis = record.split('"')[1]
+ try:
+ txtis
+ print("DNS: Found DMARC TXT record.")
+ dnsdmarcok = True
+ except NameError:
+ print("DNS: No DMARC TXT record found.")
+ dnsdmarcok = False
+except NameError:
+ print("DNS: Missing or faulty DKIM/DMARC records.")
+ dnsdmarcok = False
+ dnsdkimok = False
+except requests.exceptions.ConnectionError as err:
+ print("Connection failed.")
+ sys.exit(1)
+except requests.exceptions.HTTPError as err:
+ print(err)
+ sys.exit(1)
+
+if dnsok == True and mailok == True and dkimok == True and dnsdkimok == True and dnsdmarcok == True:
+ print("All good. No changes seem to be needed. Aborting.")
+ sys.exit(0)
+else:
+ print("Found inconsistencies:")
+ print(f"DNS OK: {dnsok} - Mail OK: {mailok} - Mail DKIM OK: {dkimok} - DNS DKIM OK: {dnsdkimok} - DNS DMARC OK: {dnsdmarcok}")
+ print("Will attempt a full repair if not cancelled within 5 seconds ...")
+ time.sleep(5)
+
+if initprimary == True:
+ print("Mail: Initializing domain ...")
+ URL = server + add + '/domain'
+ payload = {
+ "active": "1",
+ "aliases": "20",
+ "backupmx": "0",
+ "defquota": "1024",
+ "description": domain,
+ "domain": domain,
+ "mailboxes": "10",
+ "maxquota": "2048",
+ "quota": "5120",
+ "relay_all_recipients": "0",
+ "rl_frame": "s",
+ "rl_value": "10",
+ "restart_sogo": "10"
+ }
+ response = requests.post(
+ URL,
+ headers = {'accept': 'application/json', 'X-API-Key': api_key, 'Content-Type': 'application/json'},
+ json = payload,
+ )
+ data = response.json()
+ status = data[0]['type']
+ try:
+ status
+ except:
+ print("Mail Error:")
+ print(data)
+ sys.exit(1)
+ if status == 'success':
+ print("Mail: Created domain.")
+ if status == 'danger':
+ print("Mail: Failed to create domain.")
+ print(data)
+ #print(f"CREATION: {status}")
+
+if initprimary == True or dkimok == False:
+ print("Mail: Initializing DKIM ...")
+ URL = server + add + '/dkim'
+ payload = {
+ "dkim_selector": "primary",
+ "domains": domain,
+ "key_size": "2048"
+ }
+ response = requests.post(
+ URL,
+ headers = {'accept': 'application/json', 'X-API-Key': api_key, 'Content-Type': 'application/json'},
+ json = payload,
+ )
+ data = response.json()
+ status = data[0]['type']
+ try:
+ status
+ except:
+ print("Mail Error:")
+ print(data)
+ sys.exit(1)
+ if status == 'success':
+ print("Mail: Created DKIM keypair.")
+ if status == 'danger':
+ print("Mail: Failed to create DKIM keypair.")
+ print(data)
+ sys.exit(1)
+ #print(f"CREATION: {status}")
+
+URL = server + get + '/dkim/' + domain
+print("Mail: Querying DKIM ...")
+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. Unable to continue. ABORTING.")
+ sys.exit(1)
+except KeyError:
+ print("Mail: No or faulty DKIM lookup. Unable to continue. ABORTING.")
+ sys.exit(1)
+except requests.exceptions.ConnectionError as err:
+ print("Connection failed.")
+ sys.exit(1)
+except requests.exceptions.HTTPError as err:
+ print(err)
+ sys.exit(1)
+
+# PATCH
+print("DNS: Patching SPF ...")
+URL = ENDPOINT_PDNS + '/api/v1/servers/localhost/zones/' + domain + "."
+payload = {
+"rrsets": [{"name": domain + ".", "type": "TXT", "ttl": "3600", "changetype": "REPLACE", "records": [{"content": "\"v=spf1 mx a -all\"", "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("SPF: OK!")
+elif status == 422:
+ print("SPF: Failed:")
+ print(response.json())
+ sys.exit(1)
+else:
+ print("Unhandled error.")
+ print(status)
+ print(response.json())
+ sys.exit(1)
+print("DNS: Patching DMARC ...")
+URL = ENDPOINT_PDNS + '/api/v1/servers/localhost/zones/' + domain + "."
+payload = {
+"rrsets": [{"name": "_dmarc." + domain + ".", "type": "TXT", "ttl": "3600", "changetype": "REPLACE", "records": [{"content": "\"v=DMARC1; p=reject; rua=mailto:system@lysergic.dev\"", "disabled": False, "name": "._dmarc." + 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("DMARC: OK!")
+elif status == 422:
+ print("DMARC: Failed:")
+ print(response.json())
+ sys.exit(1)
+else:
+ print("Unhandled error.")
+ print(status)
+ print(response.json())
+ sys.exit(1)
+print("DNS: Patching DKIM ...")
+payload = {
+"rrsets": [{"name": selector + "._domainkey." + domain + ".", "type": "TXT", "ttl": "3600", "changetype": "REPLACE", "records": [{"content": "\""+ txtshould + "\"", "disabled": False, "name": selector + "._domainkey." + 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("DKIM: OK!")
+elif status == 422:
+ print("DKIM: Failed:")
+ print(response.json())
+ sys.exit(1)
+else:
+ print("Unhandled error.")
+ print(status)
+ print(response.json())
+ sys.exit(1)
+print("Done.")
+sys.exit(0)
+
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-22 02:16:13 +0000