add DNS check, SOA and MX

README.md

@@ -7,14 +7,17 @@ A small Flask application to check the email provided, return score and determin
 Call endpoint `GET /[email protected]` or `POST /check` with JSON `{"email": "[email protected]"}`.
 
 ```sh
-curl -H  "Content-Type: application/json" --data '{"email": "my@email.com"}' localhost:5000/check
+curl -H  "Content-Type: application/json" --data '{"email": "whatever@abaot.com"}' localhost:5000/check
 ```
 
 ```json
 {
-  "email": "[email protected]",
-  "reasons": [],
-  "score": 8,
-  "valid": true
+  "disposable": true,
+  "email": "[email protected]",
+  "reasons": [
+    "Suspicious Tempmail"
+  ],
+  "score": 0,
+  "valid": false
 }
 ```

app.py

@@ -5,6 +5,22 @@
 
 app = Flask(__name__)
 
+checks_done = 0
+
+@app.route('/health')
+def health_check():
+  return jsonify({'status': 'healthy'}), 200
+
+@app.route('/metrics')
+def metrics():
+  global checks_done
+
+  return (
+    "# HELP mail_checker_requests The number of email checks done\n"
+    "# TYPE mail_checker_requests counter\n"
+    f"mail_checker_requests {checks_done}\n"
+  ), 200, {'Content-Type': 'text/plain'}
+
 @app.route('/check', methods=['POST', 'GET'])
 def check_email():
   if request.method == 'POST':
@@ -18,7 +34,10 @@ def check_email():
 
   if not email:
     return jsonify({'error': 'Email is required'}), 400
-
+
+  global checks_done
+  checks_done += 1
+
   validator = Validator(email)
   validator.run()
 

mail_checker/const.py

@@ -0,0 +1,50 @@
+suspicious_tempmail_nameservers = [
+  ".dnsowl.com", # temp-mail.org
+]
+
+trusted_mx_servers = [
+  "alt1.aspmx.l.google.com",
+  "alt2.aspmx.l.google.com",
+  "alt3.aspmx.l.google.com",
+  "alt4.aspmx.l.google.com",
+  "aspmx.l.google.com",
+  "aspmx2.googlemail.com",
+  "aspmx3.googlemail.com",
+  "aspmx4.googlemail.com",
+  "aspmx5.googlemail.com",
+  ".mail.protection.outlook.com",
+  ".olc.protection.outlook.com",
+]
+
+# https://www.usercheck.com/providers
+tempmail_mx_servers = [
+  "mx2.den.yt",
+  "mx1.icdn.be",
+  "generator.email",
+  "tempm.com",
+  "emailfake.com",
+  "email-fake.com",
+  "mail.ggez.team",
+  "mx.cse445.com",
+  "mx1.simplelogin.co",
+  "mx2.simplelogin.co",
+  "smtp.yopmail.com",
+  "gpa.lu",
+  "mail.anonaddy.me",
+  "mail2.anonaddy.me",
+  "inbound-smtp.skiff.com",
+  "in.mail.tm",
+  "mx.discard.email",
+  "mx.moakt.com",
+  "mx.temp-mail.io",
+  "mx2.temp-mail.io",
+  "prd-smtp.10minutemail.com",
+  "loti3.papierkorb.me",
+  "us2.mx1.mailhostbox.com",
+  "us2.mx2.mailhostbox.com",
+  "us2.mx3.mailhostbox.com",
+  "mail.mailinator.com",
+  "mail2.mailinator.com",
+  "mailnesia.com",
+  "mx.1secmail.com"
+]

mail_checker/validator.py

@@ -1,7 +1,9 @@
 import logging
 import re
+import dns.resolver
 
 from .tld import tlds
+from .const import suspicious_tempmail_nameservers, trusted_mx_servers, tempmail_mx_servers
 
 logging.basicConfig(level=logging.DEBUG)
 logger = logging.getLogger(__name__)
@@ -11,6 +13,7 @@ def __init__(self, email: str):
     self.email = email
     self.score = 8
     self.reasons = list()
+    self.disposable = False # temp email
 
   @property
   def dict(self):
@@ -19,9 +22,8 @@ def dict(self):
       'valid': self.score >= 5,
       'score': self.score,
       'reasons': self.reasons,
+      'disposable': self.disposable,
     }
-    # if self.reasons:
-    #  result['reasons'] = self.reasons
     return result
 
   def is_valid_email(email):
@@ -122,6 +124,54 @@ def step_is_not_empty(self):
     if not result:
       self.penalty(10, 'Email is empty')
     return result
+
+  def step_domain_resolve_soa(self):
+    try:
+      resolve = dns.resolver.resolve(self.domain, 'SOA')
+    except dns.resolver.NXDOMAIN:
+      self.penalty(10, 'Domain does not exist')
+
+  def step_domain_resolve_suspicious_tempmail_nameservers(self):
+    """ This can be used to check if the domain is using a suspicious tempmail nameserver. """
+    try:
+      resolve = dns.resolver.resolve(self.domain, 'SOA')
+      entry = resolve.response.answer[0]
+
+      for ns in suspicious_tempmail_nameservers:
+        if ns in str(entry):
+          self.penalty(7, f'Nameserver found, suspicious tempmail: {ns}')
+          self.disposable = True
+    except dns.resolver.NXDOMAIN:
+      self.penalty(10, 'Domain does not exist')
+
+  def step_domain_resolve_mx(self):
+    try:
+      resolve = dns.resolver.resolve(self.domain, 'MX')
+    except dns.resolver.NoAnswer:
+      self.penalty(10, 'Cannot send email, no MX record found for domain')
+    except dns.resolver.NXDOMAIN:
+      self.penalty(10, 'Domain does not exist')
+    return True
+
+  def step_domain_check_mx_tempmail(self):
+    """ Check if MX belongs to known tempmail providers. """
+    try:
+      resolve = dns.resolver.resolve(self.domain, 'MX')
+      for rdata in resolve:
+        for tempmail_mx in tempmail_mx_servers:
+          if tempmail_mx in str(rdata.exchange):
+            self.penalty(7, f'Tempmail MX found: {tempmail_mx}')
+            self.disposable = True
+            return False
+        for trusted_mx in trusted_mx_servers:
+          if trusted_mx in str(rdata.exchange):
+            self.score += 2
+            # just one bump increase
+            return True
+    except dns.resolver.NoAnswer:
+      self.penalty(10, 'Cannot send email, no MX record found for domain')
+    except dns.resolver.NXDOMAIN:
+      self.penalty(10, 'Domain does not exist')
 
   def run(self):
     steps = [func for func in dir(self) if callable(getattr(self, func)) and func.startswith('step_')]

requirements.txt

@@ -1 +1,2 @@
-Flask
+Flask
+dnspython