AI-Powered Email Security: Advanced Threat Detection

mkdir -p ai-email-security/{src,data,models,logs}
cd ai-email-security
python3 -m venv venv
source venv/bin/activate
pip install --upgrade pip

pip install pandas==2.1.4 numpy==1.26.2 scikit-learn==1.3.2 tensorflow==2.15.0 nltk==3.8.1 beautifulsoup4==4.12.2 tldextract==5.1.0 email-validator==2.1.0

# src/email_features.py
"""Feature extraction from emails."""
import re
import email
from email.header import decode_header
from urllib.parse import urlparse
import tldextract
from bs4 import BeautifulSoup
import pandas as pd
import numpy as np
from typing import Dict, List, Optional
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class EmailFeatureExtractor:
    """Extracts features from email messages."""
    
    def __init__(self):
        """Initialize feature extractor."""
        self.suspicious_keywords = [
            "urgent", "verify", "suspended", "locked", "expired",
            "click here", "act now", "limited time", "winner"
        ]
        self.suspicious_tlds = [".tk", ".ml", ".ga", ".cf"]
    
    def extract_features(self, email_content: str, email_headers: Optional[Dict] = None) -> Dict:
        """
        Extract features from email.
        
        Args:
            email_content: Raw email content
            email_headers: Optional email headers dictionary
            
        Returns:
            Dictionary of extracted features
        """
        try:
            # Parse email
            msg = email.message_from_string(email_content)
            
            features = {}
            
            # Header features
            features.update(self._extract_header_features(msg, email_headers))
            
            # Content features
            features.update(self._extract_content_features(msg))
            
            # URL features
            features.update(self._extract_url_features(msg))
            
            # Behavioral features
            features.update(self._extract_behavioral_features(msg))
            
            return features
            
        except Exception as e:
            logger.error(f"Feature extraction error: {e}")
            return {}
    
    def _extract_header_features(self, msg, headers: Optional[Dict]) -> Dict:
        """Extract features from email headers."""
        features = {}
        
        # From address
        from_addr = msg.get("From", "")
        features["from_domain"] = self._extract_domain(from_addr)
        features["from_suspicious"] = self._is_suspicious_domain(features["from_domain"])
        
        # Subject
        subject = msg.get("Subject", "")
        decoded_subject = self._decode_header(subject)
        features["subject_length"] = len(decoded_subject)
        features["subject_suspicious_words"] = self._count_suspicious_words(decoded_subject)
        features["subject_has_urgency"] = any(word in decoded_subject.lower() for word in ["urgent", "immediate", "asap"])
        
        # Reply-To
        reply_to = msg.get("Reply-To", "")
        features["reply_to_different"] = reply_to and reply_to != from_addr
        
        # SPF, DKIM, DMARC
        received_spf = msg.get("Received-SPF", "")
        features["has_spf"] = "pass" in received_spf.lower()
        
        # Message ID
        message_id = msg.get("Message-ID", "")
        features["has_message_id"] = bool(message_id)
        
        return features
    
    def _extract_content_features(self, msg) -> Dict:
        """Extract features from email content."""
        features = {}
        
        # Get text content
        text_content = self._get_text_content(msg)
        html_content = self._get_html_content(msg)
        
        # Text features
        features["text_length"] = len(text_content)
        features["html_length"] = len(html_content)
        features["has_html"] = bool(html_content)
        features["suspicious_keywords_count"] = self._count_suspicious_words(text_content)
        features["link_count"] = len(re.findall(r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+])+', text_content + html_content))
        features["attachment_count"] = len([part for part in msg.walk() if part.get_content_disposition() == "attachment"])
        
        # HTML-specific features
        if html_content:
            soup = BeautifulSoup(html_content, "html.parser")
            features["html_link_count"] = len(soup.find_all("a"))
            features["html_image_count"] = len(soup.find_all("img"))
            features["has_hidden_text"] = self._has_hidden_text(soup)
        
        return features
    
    def _extract_url_features(self, msg) -> Dict:
        """Extract features from URLs in email."""
        features = {}
        
        text_content = self._get_text_content(msg) + self._get_html_content(msg)
        urls = re.findall(r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+])+', text_content)
        
        if not urls:
            features["url_count"] = 0
            features["suspicious_url_count"] = 0
            features["url_shortener_count"] = 0
            return features
        
        features["url_count"] = len(urls)
        suspicious_count = 0
        shortener_count = 0
        
        for url in urls:
            try:
                parsed = urlparse(url)
                domain = parsed.netloc
                tld_info = tldextract.extract(domain)
                
                # Check for suspicious TLDs
                if tld_info.suffix in self.suspicious_tlds:
                    suspicious_count += 1
                
                # Check for URL shorteners
                if domain in ["bit.ly", "tinyurl.com", "goo.gl", "t.co"]:
                    shortener_count += 1
                
                # Check for IP addresses in URLs
                if re.match(r'^\d+\.\d+\.\d+\.\d+$', domain):
                    suspicious_count += 1
                    
            except Exception:
                pass
        
        features["suspicious_url_count"] = suspicious_count
        features["url_shortener_count"] = shortener_count
        
        return features
    
    def _extract_behavioral_features(self, msg) -> Dict:
        """Extract behavioral features."""
        features = {}
        
        # Timing features (if available)
        date = msg.get("Date")
        features["has_date"] = bool(date)
        
        # Content type
        content_type = msg.get_content_type()
        features["content_type"] = content_type
        
        # Encoding
        encoding = msg.get("Content-Transfer-Encoding", "")
        features["has_encoding"] = bool(encoding)
        
        return features
    
    def _get_text_content(self, msg) -> str:
        """Extract plain text content from email."""
        text_parts = []
        for part in msg.walk():
            if part.get_content_type() == "text/plain":
                payload = part.get_payload(decode=True)
                if payload:
                    try:
                        text_parts.append(payload.decode("utf-8", errors="ignore"))
                    except:
                        pass
        return " ".join(text_parts)
    
    def _get_html_content(self, msg) -> str:
        """Extract HTML content from email."""
        html_parts = []
        for part in msg.walk():
            if part.get_content_type() == "text/html":
                payload = part.get_payload(decode=True)
                if payload:
                    try:
                        html_parts.append(payload.decode("utf-8", errors="ignore"))
                    except:
                        pass
        return " ".join(html_parts)
    
    def _extract_domain(self, email_addr: str) -> str:
        """Extract domain from email address."""
        match = re.search(r'@([^\s<>]+)', email_addr)
        if match:
            return match.group(1).lower()
        return ""
    
    def _is_suspicious_domain(self, domain: str) -> bool:
        """Check if domain is suspicious."""
        if not domain:
            return True
        
        tld_info = tldextract.extract(domain)
        if tld_info.suffix in self.suspicious_tlds:
            return True
        
        # Check for typosquatting patterns
        suspicious_patterns = ["paypai", "amazom", "microsft"]
        for pattern in suspicious_patterns:
            if pattern in domain.lower():
                return True
        
        return False
    
    def _count_suspicious_words(self, text: str) -> int:
        """Count suspicious keywords in text."""
        text_lower = text.lower()
        return sum(1 for keyword in self.suspicious_keywords if keyword in text_lower)
    
    def _has_hidden_text(self, soup: BeautifulSoup) -> bool:
        """Check for hidden text in HTML."""
        # Check for text with same color as background
        styles = soup.find_all(style=True)
        for style_tag in styles:
            style = style_tag.get("style", "")
            if "color:" in style and "background" in style:
                return True
        return False
    
    def _decode_header(self, header: str) -> str:
        """Decode email header."""
        try:
            decoded_parts = decode_header(header)
            decoded = []
            for part, encoding in decoded_parts:
                if isinstance(part, bytes):
                    decoded.append(part.decode(encoding or "utf-8", errors="ignore"))
                else:
                    decoded.append(part)
            return " ".join(decoded)
        except:
            return header

# src/email_classifier.py
"""ML models for email threat detection."""
import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, confusion_matrix
from sklearn.preprocessing import StandardScaler
import pickle
import logging
from pathlib import Path
from typing import Tuple

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class EmailClassifier:
    """ML classifier for email threats."""
    
    def __init__(self, model_type: str = "random_forest"):
        """
        Initialize classifier.
        
        Args:
            model_type: Type of model (random_forest, gradient_boosting)
        """
        self.model_type = model_type
        self.model = None
        self.scaler = StandardScaler()
        self.feature_names = None
        self.is_trained = False
    
    def train(self, features: pd.DataFrame, labels: pd.Series) -> None:
        """
        Train the classifier.
        
        Args:
            features: DataFrame with extracted features
            labels: Series with labels (0=benign, 1=threat)
        """
        if features.empty:
            raise ValueError("No features provided")
        
        try:
            # Store feature names
            self.feature_names = features.columns.tolist()
            
            # Split data
            X_train, X_test, y_train, y_test = train_test_split(
                features, labels, test_size=0.2, random_state=42, stratify=labels
            )
            
            # Scale features
            X_train_scaled = self.scaler.fit_transform(X_train)
            X_test_scaled = self.scaler.transform(X_test)
            
            # Train model
            if self.model_type == "random_forest":
                self.model = RandomForestClassifier(
                    n_estimators=100,
                    max_depth=10,
                    random_state=42
                )
            elif self.model_type == "gradient_boosting":
                self.model = GradientBoostingClassifier(
                    n_estimators=100,
                    max_depth=5,
                    random_state=42
                )
            else:
                raise ValueError(f"Unsupported model type: {model_type}")
            
            self.model.fit(X_train_scaled, y_train)
            
            # Evaluate
            train_score = self.model.score(X_train_scaled, y_train)
            test_score = self.model.score(X_test_scaled, y_test)
            
            logger.info(f"Training accuracy: {train_score:.3f}")
            logger.info(f"Test accuracy: {test_score:.3f}")
            
            # Print classification report
            y_pred = self.model.predict(X_test_scaled)
            print(classification_report(y_test, y_pred))
            
            self.is_trained = True
            
        except Exception as e:
            logger.error(f"Training error: {e}")
            raise
    
    def predict(self, features: pd.DataFrame) -> Tuple[np.ndarray, np.ndarray]:
        """
        Predict threats in emails.
        
        Args:
            features: DataFrame with features
            
        Returns:
            Tuple of (predictions, probabilities)
        """
        if not self.is_trained:
            raise ValueError("Model not trained")
        
        # Ensure same features
        features = features[self.feature_names]
        
        # Scale
        X_scaled = self.scaler.transform(features)
        
        # Predict
        predictions = self.model.predict(X_scaled)
        probabilities = self.model.predict_proba(X_scaled)[:, 1]
        
        return predictions, probabilities
    
    def save(self, filepath: Path) -> None:
        """Save model to file."""
        model_data = {
            "model": self.model,
            "scaler": self.scaler,
            "feature_names": self.feature_names,
            "model_type": self.model_type
        }
        with open(filepath, "wb") as f:
            pickle.dump(model_data, f)
        logger.info(f"Model saved to {filepath}")
    
    def load(self, filepath: Path) -> None:
        """Load model from file."""
        with open(filepath, "rb") as f:
            model_data = pickle.load(f)
        self.model = model_data["model"]
        self.scaler = model_data["scaler"]
        self.feature_names = model_data["feature_names"]
        self.model_type = model_data["model_type"]
        self.is_trained = True
        logger.info(f"Model loaded from {filepath}")

# src/email_detector.py
"""Real-time email threat detection."""
from src.email_features import EmailFeatureExtractor
from src.email_classifier import EmailClassifier
import pandas as pd
import logging
from typing import Dict, Optional

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class EmailThreatDetector:
    """Real-time email threat detector."""
    
    def __init__(self, classifier: EmailClassifier):
        """
        Initialize detector.
        
        Args:
            classifier: Trained email classifier
        """
        self.classifier = classifier
        self.extractor = EmailFeatureExtractor()
        self.threshold = 0.7  # Probability threshold for threat
    
    def detect(self, email_content: str, email_headers: Optional[Dict] = None) -> Dict:
        """
        Detect threats in email.
        
        Args:
            email_content: Raw email content
            email_headers: Optional email headers
            
        Returns:
            Detection result dictionary
        """
        try:
            # Extract features
            features = self.extractor.extract_features(email_content, email_headers)
            
            if not features:
                return {
                    "is_threat": False,
                    "confidence": 0.0,
                    "error": "Failed to extract features"
                }
            
            # Convert to DataFrame
            features_df = pd.DataFrame([features])
            
            # Predict
            predictions, probabilities = self.classifier.predict(features_df)
            
            is_threat = probabilities[0] >= self.threshold
            confidence = float(probabilities[0])
            
            result = {
                "is_threat": bool(is_threat),
                "confidence": confidence,
                "prediction": "threat" if is_threat else "benign",
                "features": features
            }
            
            if is_threat:
                logger.warning(f"THREAT DETECTED: confidence={confidence:.3f}")
            
            return result
            
        except Exception as e:
            logger.error(f"Detection error: {e}")
            return {
                "is_threat": False,
                "confidence": 0.0,
                "error": str(e)
            }