mailscan.py

#!/usr/bin/env python3
"""
Email Security Configuration Analyzer
Evaluates domain email security posture including SPF, DMARC, DKIM, and MX records
"""

import dns.resolver
import re
import sys
import argparse

class EmailSecurityAnalyzer:
    def __init__(self, domain: str):
        self.domain = domain.strip().lower()
        self.findings = []
        self.records = {}
        self.resolver = dns.resolver.Resolver()
        self.resolver.timeout = 5
        self.resolver.lifetime = 5

    def analyze(self):
        """Run all security checks"""
        self.check_spf()
        self.check_dmarc()
        self.check_dkim()
        self.check_mx()
        self.check_mta_sts()
        self.check_tls_rpt()
        self.check_bimi()
        return self.findings, self.records

    def check_spf(self):
        """Check SPF record"""
        try:
            answers = self.resolver.resolve(self.domain, 'TXT')
            spf_records = [str(rdata).strip('"') for rdata in answers 
                          if 'v=spf1' in str(rdata)]
            
            if not spf_records:
                self.records['SPF'] = None
                self.findings.append({
                    'category': 'SPF',
                    'risk': 'HIGH',
                    'issue': 'No SPF record found',
                    'recommendation': "Add SPF record to prevent email spoofing. Example: 'v=spf1 mx -all'"
                })
                return
            
            spf = spf_records[0]
            self.records['SPF'] = spf
            
            if len(spf_records) > 1:
                self.findings.append({
                    'category': 'SPF',
                    'risk': 'MEDIUM',
                    'issue': f"Multiple SPF records found ({len(spf_records)})",
                    'recommendation': "Consolidate into a single SPF record"
                })
            
            # Check policy
            if '+all' in spf:
                self.findings.append({
                    'category': 'SPF',
                    'risk': 'HIGH',
                    'issue': "SPF policy is '+all' (accepts email from anywhere)",
                    'recommendation': "Change to '-all' (strict) or '~all' (soft fail)"
                })
            elif '?all' in spf:
                self.findings.append({
                    'category': 'SPF',
                    'risk': 'MEDIUM',
                    'issue': "SPF policy is '?all' (neutral - no protection)",
                    'recommendation': "Change to '-all' or '~all'"
                })
            elif '~all' in spf:
                self.findings.append({
                    'category': 'SPF',
                    'risk': 'LOW',
                    'issue': "SPF policy is '~all' (soft fail - weak protection)",
                    'recommendation': "Consider '-all' for stronger enforcement"
                })
            elif '-all' in spf:
                self.findings.append({
                    'category': 'SPF',
                    'risk': 'PASS',
                    'issue': "SPF configured with strict '-all' policy",
                    'recommendation': "Keep monitoring for legitimate sender issues"
                })
            else:
                self.findings.append({
                    'category': 'SPF',
                    'risk': 'MEDIUM',
                    'issue': "SPF missing 'all' mechanism",
                    'recommendation': "Add '-all' at the end of SPF record"
                })
            
            # Check DNS lookups
            lookups = len(re.findall(r'\b(include:|a:|mx:|exists:)', spf))
            if lookups > 10:
                self.findings.append({
                    'category': 'SPF',
                    'risk': 'LOW',
                    'issue': f"SPF has {lookups} DNS lookups (limit is 10)",
                    'recommendation': "Reduce include: statements or flatten the record"
                })
                
        except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer):
            self.records['SPF'] = None
            self.findings.append({
                'category': 'SPF',
                'risk': 'HIGH',
                'issue': 'No SPF record found',
                'recommendation': "Add SPF record"
            })
        except Exception as e:
            self.records['SPF'] = f"Error: {e}"

    def check_dmarc(self):
        """Check DMARC record"""
        try:
            answers = self.resolver.resolve(f'_dmarc.{self.domain}', 'TXT')
            dmarc_records = [str(rdata).strip('"') for rdata in answers 
                           if 'v=DMARC1' in str(rdata)]
            
            if not dmarc_records:
                self.records['DMARC'] = None
                self.findings.append({
                    'category': 'DMARC',
                    'risk': 'HIGH',
                    'issue': 'No DMARC record found',
                    'recommendation': "Add DMARC at _dmarc.{domain}. Start with: v=DMARC1; p=none; rua=mailto:dmarc@{domain}"
                })
                return
            
            dmarc = dmarc_records[0]
            self.records['DMARC'] = dmarc
            
            # Check policy
            policy_match = re.search(r'p=(\w+)', dmarc)
            if not policy_match:
                self.findings.append({
                    'category': 'DMARC',
                    'risk': 'HIGH',
                    'issue': 'DMARC missing policy tag',
                    'recommendation': "Add 'p=reject' or 'p=quarantine'"
                })
                return
            
            policy = policy_match.group(1).lower()
            
            if policy == 'none':
                self.findings.append({
                    'category': 'DMARC',
                    'risk': 'LOW',
                    'issue': "DMARC policy is 'none' (monitoring only)",
                    'recommendation': "After monitoring, upgrade to 'p=quarantine' then 'p=reject'"
                })
            elif policy == 'quarantine':
                self.findings.append({
                    'category': 'DMARC',
                    'risk': 'INFORMATIONAL',
                    'issue': "DMARC policy is 'quarantine' (sends failures to spam)",
                    'recommendation': "Consider 'p=reject' for maximum protection"
                })
            elif policy == 'reject':
                self.findings.append({
                    'category': 'DMARC',
                    'risk': 'PASS',
                    'issue': "DMARC policy is 'reject' (strong protection)",
                    'recommendation': "Monitor reports for any issues"
                })
            
            # Check reporting
            if 'rua=' not in dmarc and 'ruf=' not in dmarc:
                self.findings.append({
                    'category': 'DMARC',
                    'risk': 'INFORMATIONAL',
                    'issue': 'No DMARC reporting addresses configured',
                    'recommendation': "Add 'rua=mailto:dmarc@{domain}' to receive reports"
                })
            
            # Check percentage
            pct_match = re.search(r'pct=(\d+)', dmarc)
            if pct_match and int(pct_match.group(1)) < 100:
                pct = pct_match.group(1)
                self.findings.append({
                    'category': 'DMARC',
                    'risk': 'INFORMATIONAL',
                    'issue': f"DMARC applies to only {pct}% of emails",
                    'recommendation': "Increase to pct=100 for full coverage"
                })
                
        except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer):
            self.records['DMARC'] = None
            self.findings.append({
                'category': 'DMARC',
                'risk': 'HIGH',
                'issue': 'No DMARC record found',
                'recommendation': "Add DMARC record"
            })
        except Exception as e:
            self.records['DMARC'] = f"Error: {e}"

    def check_dkim(self):
        """Check for DKIM"""
        selectors = ['default', 'selector1', 'selector2', 'google', 'k1', 's1', 'dkim']
        found = []
        
        for selector in selectors:
            try:
                self.resolver.resolve(f'{selector}._domainkey.{self.domain}', 'TXT')
                found.append(selector)
            except:
                continue
        
        if found:
            self.records['DKIM'] = f"Found selectors: {', '.join(found)}"
            self.findings.append({
                'category': 'DKIM',
                'risk': 'PASS',
                'issue': f"DKIM configured (selectors: {', '.join(found)})",
                'recommendation': "Ensure all mail servers sign with DKIM"
            })
        else:
            self.records['DKIM'] = "Not found (checked common selectors)"
            self.findings.append({
                'category': 'DKIM',
                'risk': 'LOW',
                'issue': 'No DKIM found with common selectors',
                'recommendation': "Configure DKIM signing on your mail servers"
            })

    def check_mx(self):
        """Check MX records"""
        try:
            answers = self.resolver.resolve(self.domain, 'MX')
            mx_list = sorted([(rdata.preference, str(rdata.exchange).rstrip('.')) 
                            for rdata in answers])
            
            self.records['MX'] = mx_list
            
            if len(mx_list) == 1:
                self.findings.append({
                    'category': 'MX',
                    'risk': 'INFORMATIONAL',
                    'issue': 'Only one MX record (no backup)',
                    'recommendation': "Add secondary MX for redundancy"
                })
            else:
                self.findings.append({
                    'category': 'MX',
                    'risk': 'PASS',
                    'issue': f"Found {len(mx_list)} MX records (good redundancy)",
                    'recommendation': "Monitor mail server availability"
                })
                
        except dns.resolver.NoAnswer:
            self.records['MX'] = None
            self.findings.append({
                'category': 'MX',
                'risk': 'INFORMATIONAL',
                'issue': 'No MX records (domain may not accept email)',
                'recommendation': "Configure MX if email is needed"
            })
        except Exception as e:
            self.records['MX'] = f"Error: {e}"

    def check_mta_sts(self):
        """Check MTA-STS policy"""
        try:
            # Check DNS record
            answers = self.resolver.resolve(f'_mta-sts.{self.domain}', 'TXT')
            mta_sts_records = [str(rdata).strip('"') for rdata in answers 
                              if 'v=STSv1' in str(rdata)]
            
            if mta_sts_records:
                self.records['MTA-STS'] = mta_sts_records[0]
                self.findings.append({
                    'category': 'MTA-STS',
                    'risk': 'PASS',
                    'issue': "MTA-STS policy found (enforces TLS for email delivery)",
                    'recommendation': "Ensure policy file is accessible and up to date"
                })
            else:
                self.records['MTA-STS'] = None
                self.findings.append({
                    'category': 'MTA-STS',
                    'risk': 'INFORMATIONAL',
                    'issue': 'No MTA-STS policy (optional)',
                    'recommendation': "Consider adding MTA-STS to enforce TLS encryption for incoming mail"
                })
                
        except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer):
            self.records['MTA-STS'] = None
            self.findings.append({
                'category': 'MTA-STS',
                'risk': 'INFORMATIONAL',
                'issue': 'No MTA-STS policy (optional)',
                'recommendation': "Consider adding MTA-STS to enforce TLS encryption"
            })
        except Exception as e:
            self.records['MTA-STS'] = f"Error: {e}"

    def check_tls_rpt(self):
        """Check TLS reporting"""
        try:
            answers = self.resolver.resolve(f'_smtp._tls.{self.domain}', 'TXT')
            tls_rpt_records = [str(rdata).strip('"') for rdata in answers 
                              if 'v=TLSRPTv1' in str(rdata)]
            
            if tls_rpt_records:
                self.records['TLS-RPT'] = tls_rpt_records[0]
                self.findings.append({
                    'category': 'TLS-RPT',
                    'risk': 'PASS',
                    'issue': 'TLS reporting configured',
                    'recommendation': "Monitor TLS reports for delivery issues"
                })
            else:
                self.records['TLS-RPT'] = None
                # Don't add finding - completely optional
                
        except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer):
            self.records['TLS-RPT'] = None
        except Exception as e:
            self.records['TLS-RPT'] = f"Error: {e}"

    def check_bimi(self):
        """Check BIMI record"""
        try:
            answers = self.resolver.resolve(f'default._bimi.{self.domain}', 'TXT')
            bimi_records = [str(rdata).strip('"') for rdata in answers 
                           if 'v=BIMI1' in str(rdata)]
            
            if bimi_records:
                self.records['BIMI'] = bimi_records[0]
                
                # Check if DMARC is at enforcement (required for BIMI)
                dmarc = self.records.get('DMARC', '')
                if dmarc:
                    policy_match = re.search(r'p=(quarantine|reject)', str(dmarc))
                    if policy_match:
                        self.findings.append({
                            'category': 'BIMI',
                            'risk': 'PASS',
                            'issue': 'BIMI configured with proper DMARC enforcement',
                            'recommendation': "Ensure logo SVG and VMC are accessible"
                        })
                    else:
                        self.findings.append({
                            'category': 'BIMI',
                            'risk': 'INFORMATIONAL',
                            'issue': 'BIMI configured but DMARC not at enforcement',
                            'recommendation': "BIMI requires DMARC p=quarantine or p=reject to work"
                        })
                else:
                    self.findings.append({
                        'category': 'BIMI',
                        'risk': 'INFORMATIONAL',
                        'issue': 'BIMI configured but no DMARC found',
                        'recommendation': "BIMI requires DMARC to function"
                    })
            else:
                self.records['BIMI'] = None
                # Don't add finding - completely optional branding feature
                
        except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer):
            self.records['BIMI'] = None
        except Exception as e:
            self.records['BIMI'] = f"Error: {e}"

def print_records_mode(domain, records):
    """Print records in simple format"""
    print(f"\n{'='*70}")
    print(f"EMAIL DNS RECORDS FOR: {domain}")
    print(f"{'='*70}\n")
    
    print("SPF RECORD:")
    print(f"  {records.get('SPF', 'Not found')}\n")
    
    print("DMARC RECORD:")
    print(f"  {records.get('DMARC', 'Not found')}\n")
    
    print("DKIM:")
    print(f"  {records.get('DKIM', 'Not found')}\n")
    
    print("MX RECORDS:")
    mx = records.get('MX')
    if mx:
        for priority, host in mx:
            print(f"  Priority {priority}: {host}")
    else:
        print("  Not found")
    print()
    
    print("MTA-STS:")
    print(f"  {records.get('MTA-STS', 'Not configured')}\n")
    
    print("TLS-RPT:")
    print(f"  {records.get('TLS-RPT', 'Not configured')}\n")
    
    print("BIMI:")
    print(f"  {records.get('BIMI', 'Not configured')}\n")

def print_risk_mode(domain, findings):
    """Print risk analysis"""
    print(f"\n{'='*70}")
    print(f"SECURITY RISK ANALYSIS FOR: {domain}")
    print(f"{'='*70}\n")
    
    # Group by risk level
    high = [f for f in findings if f['risk'] == 'HIGH']
    medium = [f for f in findings if f['risk'] == 'MEDIUM']
    low = [f for f in findings if f['risk'] == 'LOW']
    informational = [f for f in findings if f['risk'] == 'INFORMATIONAL']
    passed = [f for f in findings if f['risk'] == 'PASS']
    
    # Print findings by severity
    if high:
        print("🟠 HIGH RISK:")
        for f in high:
            print(f"  [{f['category']}] {f['issue']}")
            print(f"    → {f['recommendation']}\n")
    
    if medium:
        print("🟡 MEDIUM RISK:")
        for f in medium:
            print(f"  [{f['category']}] {f['issue']}")
            print(f"    → {f['recommendation']}\n")
    
    if low:
        print("🟢 LOW RISK:")
        for f in low:
            print(f"  [{f['category']}] {f['issue']}")
            print(f"    → {f['recommendation']}\n")
    
    if informational:
        print("ℹ️  INFORMATIONAL:")
        for f in informational:
            print(f"  [{f['category']}] {f['issue']}")
            print(f"    → {f['recommendation']}\n")
    
    if passed:
        print("✅ PASSING:")
        for f in passed:
            print(f"  [{f['category']}] {f['issue']}\n")
    
    # Risk score
    score = (len(high) * 50) + (len(medium) * 20) + (len(low) * 5) + (len(informational) * 1)
    
    print(f"{'='*70}")
    print(f"RISK SCORE: {score}")
    if score >= 50:
        print("⚠️  HIGH - Address security issues soon")
    elif score >= 20:
        print("⚠️  MODERATE - Some improvements recommended")
    elif score >= 5:
        print("⚠️  LOW - Minor improvements recommended")
    else:
        print("✅ GOOD - Strong email security posture")
    print(f"{'='*70}\n")

def main():
    parser = argparse.ArgumentParser(
        description='Analyze email security configuration for a domain',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python mailscan.py example.com              # Show risk analysis (default)
  python mailscan.py example.com --records    # Show DNS records only
  python mailscan.py example.com --both       # Show both risk and records
        """
    )
    parser.add_argument('domain', help='Domain to analyze')
    parser.add_argument('--records', action='store_true', 
                       help='Show DNS records only (manual analysis mode)')
    parser.add_argument('--both', action='store_true',
                       help='Show both risk analysis and DNS records')
    
    args = parser.parse_args()
    
    try:
        analyzer = EmailSecurityAnalyzer(args.domain)
        findings, records = analyzer.analyze()
        
        if args.records:
            print_records_mode(args.domain, records)
        elif args.both:
            print_risk_mode(args.domain, findings)
            print_records_mode(args.domain, records)
        else:
            print_risk_mode(args.domain, findings)
            
    except KeyboardInterrupt:
        print("\n\nAnalysis interrupted")
        sys.exit(0)
    except Exception as e:
        print(f"\nError: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()