Update app.py

app.py

@@ -1,351 +1,366 @@
 import gradio as gr
-import random
-import time
-import json
 import pandas as pd
+import asyncio
+import aiohttp
+import dns.resolver
+import smtplib
+import socket
 from typing import List, Dict, Tuple
-from dataclasses import dataclass, asdict
-from datetime import datetime
-from collections import defaultdict
+from dataclasses import dataclass
+import random
+import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
+import re
 
 @dataclass
 class EmailResult:
     email: str
     exists: bool
-    timing: int
-    priority: str
-    in_leaks: bool
-    confidence: float
-
-@dataclass
-class GenerationStats:
-    total_targets: int
-    found_in_leaks: int
-    high_priority: int
-    confirmed: int
-    generation_time: float
+    smtp_valid: bool
+    mx_valid: bool
+    disposable: bool
+    response_time: int
+    risk_score: int
 
-class ProfessionalGermanEmailGenerator:
+class RealEmailValidator:
     def __init__(self):
-        self._load_german_names()
-        self._setup_patterns()
-        self.results = []
-        self.stats = GenerationStats(0, 0, 0, 0, 0.0)
+        self.disposable_domains = self._load_disposable_domains()
+        self.common_emails = self._load_common_patterns()
         
-    def _load_german_names(self):
-        self.german_first_names = {
-            'male': [
-                ('max', 8), ('paul', 7), ('leon', 9), ('felix', 6), ('lukas', 8),
-                ('tim', 5), ('david', 7), ('elias', 4), ('ben', 9), ('jonas', 7),
-                ('luca', 6), ('finn', 5), ('henry', 3), ('noah', 4), ('emil', 4)
-            ],
-            'female': [
-                ('sofia', 6), ('emily', 5), ('nina', 4), ('jana', 5), ('amelie', 4),
-                ('anna', 8), ('lisa', 7), ('maria', 9), ('hannah', 6), ('laura', 5),
-                ('lea', 7), ('emma', 8), ('lina', 4), ('clara', 3), ('lena', 6)
-            ]
+    def _load_disposable_domains(self):
+        """Загрузка списка временных email-доменов"""
+        return {
+            'tempmail.com', '10minutemail.com', 'guerrillamail.com',
+            'mailinator.com', 'yopmail.com', 'throwawaymail.com',
+            'fakeinbox.com', 'temp-mail.org', 'trashmail.com'
         }
-        self.german_last_names = [
-            ('müller', 9), ('schmidt', 8), ('schneider', 7), ('fischer', 6),
-            ('weber', 5), ('meyer', 8), ('wagner', 6), ('becker', 5),
-            ('schulz', 7), ('hoffmann', 6), ('schäfer', 5), ('koch', 4),
-            ('richter', 4), ('klein', 7), ('wolf', 3), ('schröder', 5)
-        ]
     
-    def _setup_patterns(self):
-        self.email_patterns = [
-            ('{first}.{last}', 12), ('{first}_{last}', 10), ('{first}{last}', 8),
-            ('{last}.{first}', 6), ('{last}{first}', 5), ('{first[0]}.{last}', 9),
-            ('{first}{last}{number:02d}', 4), ('{first}.{last}{number}', 3),
-            ('{first}_{last}{number:03d}', 2), ('{last}{first[0]}{number:02d}', 3)
+    def _load_common_patterns(self):
+        """Загрузка распространенных немецких имен для генерации"""
+        german_first_names = [
+            'max', 'paul', 'leon', 'felix', 'lukas', 'tim', 'david', 'elias',
+            'ben', 'jonas', 'luca', 'finn', 'sofia', 'emily', 'nina', 'jana',
+            'amelie', 'anna', 'lisa', 'maria', 'hannah', 'laura', 'lea', 'emma'
         ]
-    
-    def _weighted_choice(self, choices):
-        total = sum(weight for _, weight in choices)
-        r = random.uniform(0, total)
-        upto = 0
-        for value, weight in choices:
-            if upto + weight >= r:
-                return value
-            upto += weight
-        return choices[-1][0]
-    
-    def _get_random_name(self, gender=None):
-        if gender is None:
-            gender = random.choice(['male', 'female'])
-        first_name = self._weighted_choice(self.german_first_names[gender])
-        last_name = self._weighted_choice(self.german_last_names)
-        return first_name, last_name, gender
-    
-    def generate_emails(self, domain: str, count: int) -> List[str]:
-        emails = []
-        used_combinations = set()
         
-        for _ in range(count * 2):
-            if len(emails) >= count:
-                break
-            pattern = self._weighted_choice(self.email_patterns)
-            first_name, last_name, gender = self._get_random_name()
-            last_name = self._process_german_chars(last_name)
-            number = None
-            if '{number}' in pattern:
-                number = self._generate_realistic_number(pattern)
-            email = self._format_email(pattern, first_name, last_name, number)
-            email = self._clean_email(email) + '@' + domain
-            if email not in used_combinations:
-                emails.append(email)
-                used_combinations.add(email)
+        german_last_names = [
+            'muller', 'schmidt', 'schneider', 'fischer', 'weber', 'meyer',
+            'wagner', 'becker', 'schulz', 'hoffmann', 'schaefer', 'koch',
+            'richter', 'klein', 'wolf', 'schroeder', 'neumann', 'schwarz'
+        ]
         
-        while len(emails) < count:
-            first_name, last_name, _ = self._get_random_name()
-            email = f"{first_name}.{last_name}@{domain}"
-            email = self._clean_email(email)
-            if email not in used_combinations:
-                emails.append(email)
-                used_combinations.add(email)
+        return german_first_names, german_last_names
+
+    def generate_german_emails(self, domain: str, count: int) -> List[str]:
+        """Генерация реалистичных немецких email-адресов"""
+        first_names, last_names = self.common_emails
+        emails = []
+        patterns = [
+            '{first}.{last}',
+            '{first}_{last}',
+            '{first}{last}',
+            '{last}.{first}',
+            '{first[0]}{last}',
+            '{first}{last}{number}',
+            '{first}.{last}{number}'
+        ]
         
-        return emails[:count]
-    
-    def _process_german_chars(self, name: str) -> str:
-        umlaut_map = {'a': 'ä', 'o': 'ö', 'u': 'ü', 'ss': 'ß'}
-        result = name
-        for original, replacement in umlaut_map.items():
-            if original in result.lower():
-                probability = 0.25 if original == 'ss' else 0.15
-                if random.random() < probability:
-                    result = result.replace(original, replacement)
-        return result
-    
-    def _generate_realistic_number(self, pattern: str) -> int:
-        if '03d' in pattern:
-            return random.randint(1, 999)
-        elif '02d' in pattern:
-            return random.randint(1, 99)
-        else:
-            return random.randint(1, 50) if random.random() < 0.7 else random.randint(51, 999)
-    
-    def _format_email(self, pattern: str, first: str, last: str, number: int) -> str:
+        for i in range(count):
+            pattern = random.choice(patterns)
+            first = random.choice(first_names)
+            last = random.choice(last_names)
+            number = random.randint(1, 999) if '{number}' in pattern else ''
+            
+            email = pattern.format(
+                first=first,
+                last=last,
+                number=number,
+                first0=first[0]
+            )
+            email = re.sub(r'[^a-zA-Z0-9._-]', '', email) + '@' + domain
+            emails.append(email.lower())
+        
+        return list(set(emails))[:count]
+
+    async def check_mx_record(self, domain: str) -> bool:
+        """Проверка MX-записей домена"""
         try:
-            if number:
-                return pattern.format(first=first, last=last, number=number, first0=first[0])
-            else:
-                return pattern.format(first=first, last=last, first0=first[0])
+            answers = dns.resolver.resolve(domain, 'MX')
+            return len(answers) > 0
         except:
-            return f"{first}.{last}"
-    
-    def _clean_email(self, email: str) -> str:
-        replacements = {'ä': 'ae', 'ö': 'oe', 'ü': 'ue', 'ß': 'ss'}
-        for umlaut, replacement in replacements.items():
-            email = email.replace(umlaut, replacement)
-        email = email.replace('..', '.').replace('__', '_')
-        return email.lower().strip()
-    
-    def simulate_sophisticated_check(self, email: str) -> Tuple[bool, int, float]:
+            return False
+
+    async def smtp_verify(self, email: str) -> Tuple[bool, int]:
+        """Проверка email через SMTP"""
         start_time = time.time()
-        local_part = email.split('@')[0]
-        delay_factors = self._calculate_delay_factors(email, local_part)
-        total_delay = delay_factors['base_delay'] * delay_factors['total_factor']
-        time.sleep(total_delay)
-        timing = int((time.time() - start_time) * 1000)
-        exists, confidence = self._determine_existence(email, local_part, delay_factors)
-        return exists, timing, confidence
-    
-    def _calculate_delay_factors(self, email: str, local_part: str) -> Dict:
-        base_delay = random.uniform(0.15, 0.25)
-        factors = {
-            'domain': 1.0, 'complexity': 1.0,
-            'network': random.uniform(0.9, 1.1), 'server_load': random.uniform(0.8, 1.2)
-        }
         domain = email.split('@')[1]
-        domain_delays = {
-            'freenet.de': (0.8, 1.1), 'gmx.de': (0.9, 1.3),
-            'web.de': (0.7, 1.0), 't-online.de': (1.0, 1.4)
-        }
-        factors['domain'] = random.uniform(*domain_delays.get(domain, (0.8, 1.2)))
-        if len(local_part) > 15:
-            factors['complexity'] *= 1.1
-        if local_part.count('.') > 1:
-            factors['complexity'] *= 1.05
-        total_factor = factors['domain'] * factors['complexity'] * factors['network'] * factors['server_load']
-        return {'base_delay': base_delay, 'total_factor': total_factor, 'factors': factors}
-    
-    def _determine_existence(self, email: str, local_part: str, delay_factors: Dict) -> Tuple[bool, float]:
-        email_hash = hash(email) % 1000
-        base_probability = 28
-        adjustments = 0
-        if '.' in local_part and not any(char.isdigit() for char in local_part):
-            adjustments += 12
-        elif '_' in local_part:
-            adjustments += 8
-        if len(local_part) < 10:
-            adjustments += 5
-        popular_names = ['max', 'paul', 'anna', 'lisa', 'tim']
-        if any(name in local_part.lower() for name in popular_names):
-            adjustments += 7
-        final_probability = min(70, base_probability + adjustments)
-        exists = email_hash < final_probability
-        confidence = 0.5
-        if exists:
-            if adjustments > 15:
-                confidence = 0.8
-            elif adjustments > 10:
-                confidence = 0.7
-            else:
-                confidence = 0.6
-        else:
-            if adjustments < 5:
-                confidence = 0.7
-            else:
-                confidence = 0.6
-        confidence += random.uniform(-0.1, 0.1)
-        confidence = max(0.3, min(0.9, confidence))
-        return exists, confidence
-    
-    def classify_email(self, email: str, exists: bool, timing: int, confidence: float) -> EmailResult:
-        in_leaks = False
-        priority = "low"
-        if exists:
-            leak_probability = 35
-            email_hash = hash(email) % 100
-            in_leaks = email_hash < leak_probability
-            if timing < 400:
-                priority = "high"
-            elif timing < 700 and in_leaks:
-                priority = "high"
-            elif timing < 600:
-                priority = "medium"
-        return EmailResult(email=email, exists=exists, timing=timing, priority=priority, in_leaks=in_leaks, confidence=confidence)
-    
-    def generate_and_check(self, domain: str, count: int) -> Tuple[pd.DataFrame, Dict]:
-        start_time = time.time()
-        emails = self.generate_emails(domain, count)
-        self.results = []
-        stats_counter = defaultdict(int)
-        
-        for email in emails:
-            exists, timing, confidence = self.simulate_sophisticated_check(email)
-            result = self.classify_email(email, exists, timing, confidence)
-            self.results.append(result)
-            stats_counter['total'] += 1
-            if exists:
-                stats_counter['exists'] += 1
-            if result.in_leaks:
-                stats_counter['leaks'] += 1
-            if result.priority == "high":
-                stats_counter['high_priority'] += 1
-            if exists and result.priority == "high":
-                stats_counter['confirmed'] += 1
         
-        generation_time = time.time() - start_time
-        self.stats = GenerationStats(
-            total_targets=stats_counter['total'],
-            found_in_leaks=stats_counter['leaks'],
-            high_priority=stats_counter['high_priority'],
-            confirmed=stats_counter['confirmed'],
-            generation_time=generation_time
-        )
-        
-        # Convert to DataFrame for Gradio display
-        df_data = []
-        for result in self.results:
-            status_icon = "✅" if result.exists else "❌"
-            leaks_icon = "🔓" if result.in_leaks else ""
-            priority_icon = "🔴" if result.priority == "high" else "🟡" if result.priority == "medium" else "🟢"
+        try:
+            # Получаем MX-записи
+            mx_records = []
+            try:
+                answers = dns.resolver.resolve(domain, 'MX')
+                mx_records = sorted([(r.preference, str(r.exchange)) for r in answers])
+            except:
+                return False, 0
+
+            if not mx_records:
+                return False, 0
+
+            # Пробуем подключиться к SMTP-серверу
+            for preference, mx in mx_records[:3]:  # Проверяем первые 3 MX
+                try:
+                    with smtplib.SMTP(timeout=10) as server:
+                        server.set_debuglevel(0)
+                        server.connect(mx)
+                        server.helo()
+                        server.mail('test@example.com')
+                        code, message = server.rcpt(email)
+                        response_time = int((time.time() - start_time) * 1000)
+                        
+                        # Коды ответов, указывающие на существование email
+                        if code in [250, 251]:
+                            return True, response_time
+                        else:
+                            return False, response_time
+                            
+                except (smtplib.SMTPServerDisconnected, smtplib.SMTPConnectError,
+                       socket.timeout, socket.gaierror):
+                    continue
+                    
+        except Exception as e:
+            print(f"SMTP error for {email}: {e}")
             
-            df_data.append({
-                "Email": result.email,
-                "Status": f"{status_icon} {'Exists' if result.exists else 'Not Found'}",
-                "Timing (ms)": result.timing,
-                "Priority": f"{priority_icon} {result.priority.title()}",
-                "In Leaks": f"{leaks_icon} {'Yes' if result.in_leaks else 'No'}",
-                "Confidence": f"{result.confidence:.1%}"
-            })
+        return False, int((time.time() - start_time) * 1000)
+
+    def check_disposable(self, email: str) -> bool:
+        """Проверка на временный email"""
+        domain = email.split('@')[1].lower()
+        return domain in self.disposable_domains
+
+    def calculate_risk_score(self, email: str, exists: bool, disposable: bool) -> int:
+        """Расчет оценки риска"""
+        score = 0
         
-        df = pd.DataFrame(df_data)
-        stats_dict = asdict(self.stats)
+        if not exists:
+            score += 80
+        if disposable:
+            score += 90
+            
+        # Проверка паттернов
+        local_part = email.split('@')[0]
+        if re.match(r'^[a-z]+\.[a-z]+$', local_part):  # name.surname
+            score -= 20
+        elif re.match(r'^[a-z]+[0-9]+$', local_part):  # name123
+            score += 10
+        elif len(local_part) < 5:
+            score += 30
+            
+        return min(100, max(0, score))
+
+    async def validate_single_email(self, email: str) -> EmailResult:
+        """Валидация одного email-адреса"""
+        try:
+            disposable = self.check_disposable(email)
+            mx_valid = await self.check_mx_record(email.split('@')[1])
+            smtp_valid, response_time = await self.smtp_verify(email)
+            
+            exists = mx_valid and smtp_valid
+            risk_score = self.calculate_risk_score(email, exists, disposable)
+            
+            return EmailResult(
+                email=email,
+                exists=exists,
+                smtp_valid=smtp_valid,
+                mx_valid=mx_valid,
+                disposable=disposable,
+                response_time=response_time,
+                risk_score=risk_score
+            )
+            
+        except Exception as e:
+            print(f"Validation error for {email}: {e}")
+            return EmailResult(
+                email=email,
+                exists=False,
+                smtp_valid=False,
+                mx_valid=False,
+                disposable=False,
+                response_time=0,
+                risk_score=100
+            )
+
+    async def validate_emails_batch(self, emails: List[str]) -> List[EmailResult]:
+        """Пакетная валидация email-адресов"""
+        tasks = [self.validate_single_email(email) for email in emails]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
         
-        return df, stats_dict
+        valid_results = []
+        for result in results:
+            if isinstance(result, EmailResult):
+                valid_results.append(result)
+            else:
+                print(f"Error in validation: {result}")
+                
+        return valid_results
 
-# Initialize the generator
-generator = ProfessionalGermanEmailGenerator()
+# Инициализация валидатора
+validator = RealEmailValidator()
 
-def generate_emails_interface(domain, count):
-    """Gradio interface function"""
-    count = int(count)
-    if count > 1000:
-        count = 1000
+async def validate_emails_interface(domain, count, use_custom_emails, custom_emails_text):
+    """Интерфейс для Gradio"""
+    # Генерация или использование кастомных email-адресов
+    if use_custom_emails and custom_emails_text:
+        emails = [email.strip() for email in custom_emails_text.split('\n') if email.strip()]
+    else:
+        emails = validator.generate_german_emails(domain, count)
+    
+    # Валидация
+    results = await validator.validate_emails_batch(emails)
     
-    df, stats = generator.generate_and_check(domain, count)
+    # Подготовка данных для таблицы
+    df_data = []
+    stats = {
+        'total': len(results),
+        'valid': 0,
+        'invalid': 0,
+        'disposable': 0,
+        'high_risk': 0
+    }
+    
+    for result in results:
+        status = "✅ Valid" if result.exists else "❌ Invalid"
+        mx_status = "✅" if result.mx_valid else "❌"
+        smtp_status = "✅" if result.smtp_valid else "❌"
+        disposable_status = "⚠️ Yes" if result.disposable else "✅ No"
+        
+        risk_color = "🔴" if result.risk_score > 70 else "🟡" if result.risk_score > 30 else "🟢"
+        
+        df_data.append({
+            "Email": result.email,
+            "Status": status,
+            "MX Record": mx_status,
+            "SMTP Check": smtp_status,
+            "Disposable": disposable_status,
+            "Response Time": f"{result.response_time}ms",
+            "Risk Score": f"{risk_color} {result.risk_score}%"
+        })
+        
+        # Статистика
+        if result.exists:
+            stats['valid'] += 1
+        else:
+            stats['invalid'] += 1
+            
+        if result.disposable:
+            stats['disposable'] += 1
+            
+        if result.risk_score > 70:
+            stats['high_risk'] += 1
     
-    # Format statistics for display
+    # Форматирование статистики
     stats_text = f"""
-    ## 📊 Generation Statistics
+    ## 📊 Validation Statistics
     
-    | Metric | Value |
-    |--------|-------|
-    | Total Targets | {stats['total_targets']} |
-    | Found in Leaks | {stats['found_in_leaks']} |
-    | High Priority | {stats['high_priority']} |
-    | Confirmed | {stats['confirmed']} |
-    | Generation Time | {stats['generation_time']:.2f}s |
+    | Metric | Value | Percentage |
+    |--------|-------|------------|
+    | Total Emails | {stats['total']} | 100% |
+    | Valid Emails | {stats['valid']} | {stats['valid']/stats['total']*100:.1f}% |
+    | Invalid Emails | {stats['invalid']} | {stats['invalid']/stats['total']*100:.1f}% |
+    | Disposable Emails | {stats['disposable']} | {stats['disposable']/stats['total']*100:.1f}% |
+    | High Risk Emails | {stats['high_risk']} | {stats['high_risk']/stats['total']*100:.1f}% |
     """
     
+    df = pd.DataFrame(df_data)
     return df, stats_text
 
-# Create Gradio interface
-with gr.Blocks(theme=gr.themes.Soft(), title="German Email Generator") as demo:
+def validate_emails_sync(domain, count, use_custom_emails, custom_emails_text):
+    """Синхронная обертка для асинхронной функции"""
+    return asyncio.run(validate_emails_interface(domain, count, use_custom_emails, custom_emails_text))
+
+# Создание интерфейса Gradio
+with gr.Blocks(theme=gr.themes.Soft(), title="Real Email Validator") as demo:
     gr.Markdown("""
-    # 🇩🇪 German Email Address Generator
-    *Generate realistic German email addresses for security research and testing purposes*
+    # 🔍 Real Email Address Validator
+    *Professional tool for real email validation and verification*
     """)
     
     with gr.Row():
         with gr.Column(scale=1):
-            domain = gr.Dropdown(
-                choices=["freenet.de", "gmx.de", "web.de", "t-online.de", "gmail.com"],
-                value="freenet.de",
-                label="Target Domain"
-            )
-            count = gr.Slider(
-                minimum=1,
-                maximum=200,
-                value=20,
-                step=1,
-                label="Number of Emails to Generate"
-            )
-            generate_btn = gr.Button("🎯 Generate Emails", variant="primary")
+            with gr.Group():
+                gr.Markdown("### 🎯 Generation Settings")
+                domain = gr.Dropdown(
+                    choices=["gmail.com", "yahoo.com", "hotmail.com", "outlook.com", 
+                            "web.de", "gmx.de", "freenet.de", "t-online.de"],
+                    value="gmail.com",
+                    label="Target Domain"
+                )
+                count = gr.Slider(
+                    minimum=1,
+                    maximum=50,
+                    value=10,
+                    step=1,
+                    label="Number of Emails to Generate"
+                )
+            
+            with gr.Group():
+                gr.Markdown("### 📧 Custom Emails")
+                use_custom_emails = gr.Checkbox(
+                    label="Use custom email list",
+                    value=False
+                )
+                custom_emails_text = gr.Textbox(
+                    label="Enter emails (one per line)",
+                    lines=5,
+                    placeholder="user1@domain.com\nuser2@domain.com\n...",
+                    visible=False
+                )
+            
+            validate_btn = gr.Button("🔍 Validate Emails", variant="primary", size="lg")
         
         with gr.Column(scale=2):
             stats_output = gr.Markdown(label="Statistics")
     
     with gr.Row():
         results_table = gr.Dataframe(
-            headers=["Email", "Status", "Timing (ms)", "Priority", "In Leaks", "Confidence"],
-            datatype=["str", "str", "number", "str", "str", "str"],
-            label="Generated Email Addresses",
-            interactive=False
+            headers=["Email", "Status", "MX Record", "SMTP Check", "Disposable", "Response Time", "Risk Score"],
+            datatype=["str", "str", "str", "str", "str", "str", "str"],
+            label="Email Validation Results",
+            interactive=False,
+            wrap=True
         )
     
     gr.Markdown("""
-    ### 🔍 How it works:
-    - **Realistic Generation**: Uses common German names and email patterns
-    - **Smart Validation**: Simulates email existence checking with timing analysis
-    - **Priority Classification**: Identifies high-value targets based on response patterns
-    - **Leak Detection**: Flags emails potentially found in data breaches
+    ### 🔧 Validation Methods:
+    - **MX Record Check**: Verifies domain mail server configuration
+    - **SMTP Verification**: Real SMTP handshake to check email existence
+    - **Disposable Detection**: Identifies temporary email services
+    - **Risk Assessment**: Comprehensive risk scoring based on multiple factors
     
     ### ⚠️ Important Notes:
-    - This tool is for **educational and security research purposes only**
-    - All email checks are **simulated** for demonstration
-    - Respect privacy and applicable laws when using this tool
+    - This tool performs **real SMTP verification** - use responsibly
+    - Some mail servers may block verification attempts
+    - Results may vary based on server configurations
+    - Respect rate limits and terms of service
+    - For educational and authorized testing only
     """)
     
-    # Connect the button
-    generate_btn.click(
-        fn=generate_emails_interface,
-        inputs=[domain, count],
+    # Обработчики видимости
+    def toggle_custom_emails(use_custom):
+        return gr.Textbox(visible=use_custom)
+    
+    use_custom_emails.change(
+        fn=toggle_custom_emails,
+        inputs=[use_custom_emails],
+        outputs=[custom_emails_text]
+    )
+    
+    # Основной обработчик
+    validate_btn.click(
+        fn=validate_emails_sync,
+        inputs=[domain, count, use_custom_emails, custom_emails_text],
         outputs=[results_table, stats_output]
     )
 
-# For Hugging Face Spaces
 if __name__ == "__main__":
     demo.launch(share=True)