Init mxme.py (Fullchain Mail/DNS script)

Signed-off-by: Georg <georg@lysergic.dev>

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)
+