add validate_spf

validate_spf/get_domains.py

@@ -0,0 +1,17 @@
+from hobo.environment.models import Hobo, Variable
+import json
+
+domains = {}
+
+for hobo in Hobo.objects.all():
+    try:
+        address = Variable.objects.get(name='default_from_email')
+    except Variable.DoesNotExist:
+        continue
+    if address and address.value != '':
+        domain = address.value.split('@')[-1]
+        domains.setdefault(domain, [])
+        domains[domain].append((address.value, hobo.slug))
+
+if domains:
+    print(json.dumps(domains))

validate_spf/validate_spf.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python3
+import json
+import pprint
+import subprocess
+import dns.resolver
+import dns.exception
+import sys
+
+allowed_records = [b'include:entrouvert.com']
+
+
+class ValidationError(Exception):
+    pass
+
+
+def validate_email_spf(email_domain, strict=False):
+    txt_records = []
+    try:
+        for txt_record in dns.resolver.query(email_domain, 'TXT'):
+            txt_records.extend(txt_record.strings)
+    except dns.exception.DNSException:
+        pass
+    spf_records = [x for x in txt_records if x.startswith(b'v=spf1 ')]
+    if not strict and not spf_records:
+        return
+    for spf_record in spf_records:
+        if b'+all' in spf_record:
+            return
+        for allowed_record in allowed_records:
+            if allowed_record in spf_record:
+                return
+    raise ValidationError(_('No suitable SPF record found for %s') % email_domain)
+
+
+def get_domains(host):
+    domains = {}
+    script = 'get_domains.py'
+    subprocess.run('scp %s %s:' % (script, host), shell=True, check=True)
+    r = subprocess.check_output('ssh %s sudo -u hobo hobo-manage tenant_command runscript --all-tenants %s' %
+                                (host, script), shell=True)
+    for l in r.splitlines():
+        d = json.loads(l)
+        for k, v in d.items():
+            domains.setdefault(k, [])
+            domains[k].append(v)
+    return domains
+
+
+domains = get_domains(sys.argv[1])
+
+pprint.pprint(domains)
+for domain in domains.keys():
+    try:
+        validate_email_spf(domain, strict=True)
+    except ValidationError:
+        print('domain %s does not have %s' % (domain, allowed_records))