How to Detect Phishing Emails with Headers

#!/usr/bin/env python3
# analyze-email-headers.py

import email
import sys
import re

def analyze(filepath):
    with open(filepath, 'r', errors='replace') as f:
        msg = email.message_from_file(f)

    print("=== SENDER ANALYSIS ===")
    print(f"From:        {msg.get('From', 'NOT PRESENT')}")
    print(f"Reply-To:    {msg.get('Reply-To', 'not set')}")
    print(f"Return-Path: {msg.get('Return-Path', 'NOT PRESENT')}")

    print("\n=== AUTHENTICATION ===")
    auth = msg.get('Authentication-Results', '')
    for proto in ['spf', 'dkim', 'dmarc']:
        match = re.search(rf'{proto}=(\S+)', auth, re.IGNORECASE)
        result = match.group(1) if match else 'NOT FOUND'
        flag = "OK" if result == 'pass' else "FAIL"
        print(f"{proto.upper()}: {result} [{flag}]")

    print("\n=== RECEIVED CHAIN (bottom to top = first to last hop) ===")
    received = msg.get_all('Received', [])
    for i, hop in enumerate(reversed(received)):
        # Extract IP from "from X (X [IP])"
        ip_match = re.search(r'\[(\d+\.\d+\.\d+\.\d+)\]', hop)
        ip = ip_match.group(1) if ip_match else 'unknown'
        print(f"Hop {i+1}: {ip}")
        print(f"  {hop[:120].strip()}")

    # Red flags
    print("\n=== RED FLAGS ===")
    from_domain = re.search(r'@([^>]+)', msg.get('From', ''))
    rt_domain = re.search(r'@([^>]+)', msg.get('Return-Path', ''))
    if from_domain and rt_domain:
        fd = from_domain.group(1).rstrip('>')
        rd = rt_domain.group(1).rstrip('>')
        if fd.lower() != rd.lower():
            print(f"! From domain ({fd}) != Return-Path domain ({rd})")
    if msg.get('Reply-To'):
        print(f"! Reply-To is set: {msg.get('Reply-To')}")
    if 'dkim=fail' in auth.lower() or 'dkim=none' in auth.lower():
        print("! DKIM failed or missing")
    if 'dmarc=fail' in auth.lower():
        print("! DMARC failed")

if __name__ == '__main__':
    analyze(sys.argv[1])

#!/usr/bin/env python3
"""Build searchable database of phishing attempts."""

import sqlite3
import email
import hashlib
from datetime import datetime

class PhishingDatabase:
    def __init__(self, db_file="phishing.db"):
        self.conn = sqlite3.connect(db_file)
        self.init_schema()

    def init_schema(self):
        """Create database tables."""
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS phishing_emails (
                id INTEGER PRIMARY KEY,
                timestamp DATETIME,
                from_domain TEXT,
                claimed_sender TEXT,
                real_sender TEXT,
                subject TEXT,
                sender_ip TEXT,
                sender_country TEXT,
                spf_result TEXT,
                dkim_result TEXT,
                dmarc_result TEXT,
                phishing_type TEXT,
                content_hash TEXT UNIQUE
            )
        """)

        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS indicators (
                email_id INTEGER,
                indicator_type TEXT,
                indicator_value TEXT,
                severity TEXT,
                FOREIGN KEY(email_id) REFERENCES phishing_emails(id)
            )
        """)
        self.conn.commit()

    def analyze_email(self, filepath: str) -> dict:
        """Parse email and extract phishing indicators."""
        with open(filepath, 'rb') as f:
            msg = email.message_from_bytes(f.read())

        analysis = {
            "from_domain": msg.get('From', '').split('@')[-1].rstrip('>'),
            "claimed_sender": msg.get('From', ''),
            "real_sender": msg.get('Return-Path', '').split('@')[-1].rstrip('>'),
            "subject": msg.get('Subject', ''),
            "spf": msg.get('Received-SPF', 'none'),
            "dkim": msg.get('DKIM-Signature', 'none'),
            "dmarc": msg.get('Authentication-Results', 'none'),
            "content_hash": hashlib.sha256(str(msg).encode()).hexdigest()
        }

        # Identify indicators
        indicators = []
        if analysis["from_domain"] != analysis["real_sender"]:
            indicators.append({
                "type": "Domain Mismatch",
                "value": f"{analysis['from_domain']} != {analysis['real_sender']}",
                "severity": "HIGH"
            })

        if "fail" in analysis["spf"].lower():
            indicators.append({
                "type": "SPF Failure",
                "value": analysis["spf"],
                "severity": "HIGH"
            })

        return {
            "analysis": analysis,
            "indicators": indicators
        }

    def record_email(self, filepath: str, phishing_type: str = "unknown"):
        """Record analyzed email in database."""
        result = self.analyze_email(filepath)
        analysis = result["analysis"]

        try:
            cursor = self.conn.execute("""
                INSERT INTO phishing_emails
                (timestamp, from_domain, claimed_sender, real_sender, subject,
                 spf_result, dkim_result, dmarc_result, phishing_type, content_hash)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                datetime.now(),
                analysis["from_domain"],
                analysis["claimed_sender"],
                analysis["real_sender"],
                analysis["subject"],
                analysis["spf"],
                analysis["dkim"],
                analysis["dmarc"],
                phishing_type,
                analysis["content_hash"]
            ))

            email_id = cursor.lastrowid

            # Record indicators
            for indicator in result["indicators"]:
                self.conn.execute("""
                    INSERT INTO indicators
                    (email_id, indicator_type, indicator_value, severity)
                    VALUES (?, ?, ?, ?)
                """, (
                    email_id,
                    indicator["type"],
                    indicator["value"],
                    indicator["severity"]
                ))

            self.conn.commit()
            return email_id
        except sqlite3.IntegrityError:
            print("Email already in database (duplicate)")
            return None

# Usage
db = PhishingDatabase()
email_id = db.record_email("suspicious.eml", "spoofing")
print(f"Recorded email ID: {email_id}")