Disposable Email Detection: How to Block 250,000+ Fake Signups and Save $40K Monthly

// Frontend disposable email detection
import { DisposableEmailDetector } from '@email-check/app-js-sdk';

const detector = new DisposableEmailDetector({
  apiKey: 'your-api-key',
  blockNewDomains: true, // Block newly registered domains
  checkPatterns: true,   // AI-powered pattern detection
  updateInterval: 3600000 // Update blacklist every hour
});

// Detect disposable emails during sign-up
document.getElementById('email').addEventListener('blur', async (e) => {
  const result = await detector.detect(e.target.value);

  if (result.isDisposable) {
    showError('Disposable emails are not allowed');
    document.getElementById('signup-form').disabled = true;

    // Show user-friendly message
    showDisposableWarning(result.reason);
  } else if (result.isRisky) {
    showWarning('This email address looks suspicious');
    // Require additional verification
    enableAdditionalVerification();
  } else {
    enableForm();
  }
});

// Bulk check existing user database
async function auditExistingUsers() {
  const emails = await fetchUserEmails();
  const results = await detector.bulkCheck(emails);

  const disposableUsers = results.filter(r => r.isDisposable);
  console.log(`Found ${disposableUsers.length} disposable email users`);

  // Flag accounts for review or suspension
  await flagSuspiciousAccounts(disposableUsers);
}

// Node.js/Express backend disposable email detection
const express = require('express');
const rateLimit = require('express-rate-limit');

const app = express();

// Rate limiting to prevent brute force attacks
const signupLimiter = rateLimit({
  windowMs: 15 * 60 * 1000, // 15 minutes
  max: 5, // limit each IP to 5 signups per windowMs
  message: 'Too many signup attempts from this IP'
});

app.use(express.json());
app.use('/api/signup', signupLimiter);

app.post('/api/signup', async (req, res) => {
  const { email, name, company } = req.body;

  try {
    // Check for disposable email
    const disposableCheck = await fetch('https://api.email-check.app/v1/disposable-check', {
      method: 'POST',
      headers: {
        'Authorization': 'Bearer YOUR_API_KEY',
        'Content-Type': 'application/json'
      },
      body: JSON.stringify({
        email: email,
        includeNewDomains: true,
        checkPatterns: true,
        riskScore: true
      })
    });

    const result = await disposableCheck.json();

    if (result.isDisposable) {
      return res.status(400).json({
        error: 'Disposable email detected',
        reason: result.reason,
        blocked: true
      });
    }

    if (result.riskScore > 0.8) {
      return res.status(400).json({
        error: 'High-risk email detected',
        reason: 'Email pattern matches known fraud attempts',
        requiresManualReview: true
      });
    }

    // Email is safe - proceed with signup
    const user = await createUser({ email, name, company });

    res.json({
      success: true,
      userId: user.id,
      message: 'Account created successfully'
    });

  } catch (error) {
    console.error('Disposable email check failed:', error);
    res.status(500).json({ error: 'Validation service unavailable' });
  }
});

// Scheduled task to audit existing users
setInterval(async () => {
  console.log('Running disposable email audit...');
  const suspiciousAccounts = await auditUserDatabase();

  if (suspiciousAccounts.length > 0) {
    await flagForReview(suspiciousAccounts);
    console.log(`Flagged ${suspiciousAccounts.length} accounts for review`);
  }
}, 24 * 60 * 60 * 1000); // Run daily

app.listen(3000);

import pandas as pd
import requests
import asyncio
from datetime import datetime, timedelta

class DisposableEmailAuditor:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = 'https://api.email-check.app/v1'
        self.blacklist_cache = {}

    def check_disposable_email(self, email):
        """Check if email is from disposable service"""
        try:
            response = requests.post(
                f'{self.base_url}/disposable-check',
                headers={'Authorization': f'Bearer {self.api_key}'},
                json={
                    'email': email,
                    'include_new_domains': True,
                    'check_patterns': True,
                    'risk_scoring': True
                },
                timeout=5
            )
            return response.json()
        except Exception as e:
            return {'email': email, 'error': str(e), 'isDisposable': False}

    def audit_database(self, db_connection, table_name='users', email_column='email'):
        """Audit entire user database for disposable emails"""
        # Load all users from database
        users_df = pd.read_sql(f"SELECT id, {email_column}, created_at FROM {table_name}", db_connection)

        print(f"Auditing {len(users_df)} users for disposable emails...")

        # Check emails in batches
        batch_size = 100
        disposable_users = []

        for i in range(0, len(users_df), batch_size):
            batch = users_df.iloc[i:i+batch_size]

            for _, user in batch.iterrows():
                result = self.check_disposable_email(user[email_column])

                if result.get('isDisposable', False):
                    disposable_users.append({
                        'user_id': user['id'],
                        'email': user[email_column],
                        'reason': result.get('reason', 'Disposable email detected'),
                        'risk_score': result.get('riskScore', 1.0),
                        'created_at': user['created_at']
                    })

            # Progress indicator
            if i % 1000 == 0:
                print(f"Processed {i}/{len(users_df)} users...")

        return disposable_users

    def generate_audit_report(self, disposable_users):
        """Generate detailed audit report"""
        if not disposable_users:
            return "No disposable emails found in database."

        total_fake_accounts = len(disposable_users)
        avg_risk_score = sum(user['riskScore'] for user in disposable_users) / total_fake_accounts
        monthly_savings = round(total_fake_accounts * 0.16, 2)

        return "Disposable Email Audit Report - Action required: Review and suspend flagged accounts."

    def flag_accounts_in_database(self, disposable_users, db_connection, flag_table='flagged_accounts'):
        """Flag suspicious accounts in database"""
        if not disposable_users:
            return 0

        # Create flagged accounts table if it doesn't exist
        db_connection.execute(f"""
            CREATE TABLE IF NOT EXISTS {flag_table} (
                user_id INTEGER PRIMARY KEY,
                email TEXT,
                reason TEXT,
                risk_score REAL,
                flagged_at TIMESTAMP,
                status TEXT DEFAULT 'pending_review'
            )
        """)

        # Insert flagged accounts
        flagged_count = 0
        for user in disposable_users:
            try:
                db_connection.execute(f"""
                    INSERT OR REPLACE INTO {flag_table}
                    (user_id, email, reason, risk_score, flagged_at)
                    VALUES (?, ?, ?, ?, ?)
                """, (
                    user['user_id'],
                    user['email'],
                    user['reason'],
                    user['risk_score'],
                    datetime.now()
                ))
                flagged_count += 1
            except Exception as e:
                print(f"Error flagging user {user['user_id']}: {e}")

        db_connection.commit()
        return flagged_count

# Usage example
auditor = DisposableEmailAuditor('your-api-key-here')

# Connect to your database
import sqlite3
conn = sqlite3.connect('your_database.db')

# Run audit
disposable_users = auditor.audit_database(conn, 'users', 'email')

# Generate report
report = auditor.generate_audit_report(disposable_users)
print(report)

# Save report to file
with open('disposable_email_audit_report.md', 'w') as f:
    f.write(report)

# Flag suspicious accounts
flagged_count = auditor.flag_accounts_in_database(disposable_users, conn)
print(f"Flagged {flagged_count} accounts for review")