Phishing Authentication: What No One Tells You

// FIDO2 Passkey Registration Implementation with @simplewebauthn/server v9.0.0
// Dependencies: npm install @simplewebauthn/server express uuid
import { generateRegistrationOptions, verifyRegistrationResponse } from '@simplewebauthn/server';
import { Express, Request, Response, NextFunction } from 'express';
import { v4 as uuidv4 } from 'uuid';
import { UserService } from './user-service'; // Assume this handles user CRUD

// Configure WebAuthn relying party (your app's domain)
const rpName = 'Acme Corp Auth';
const rpID = 'auth.acme.example'; // Must match your domain, no port for production
const origin = `https://${rpID}`; // For local dev, add port e.g., https://localhost:3000

export function registerPasskeyRoutes(app: Express) {
  // 1. Initiate passkey registration for a logged-in user
  app.post('/auth/passkey/register/init', async (req: Request, res: Response, next: NextFunction) => {
    try {
      const { userId } = req.session; // Assume session is authenticated
      if (!userId) {
        return res.status(401).json({ error: 'Unauthenticated' });
      }

      const user = await UserService.findById(userId);
      if (!user) {
        return res.status(404).json({ error: 'User not found' });
      }

      // Generate registration options with security constraints
      const options = generateRegistrationOptions({
        rpName,
        rpID,
        userID: user.id,
        userName: user.email,
        userDisplayName: user.fullName,
        attestationType: 'none', // Skip attestation for most use cases, reduces complexity
        authenticatorSelection: {
          authenticatorAttachment: 'platform', // Prefer platform authenticators (TouchID, Windows Hello)
          requireResidentKey: false,
          userVerification: 'preferred',
        },
        supportedAlgorithmIDs: [-7, -257], // ES256 and RS256 algorithms
      });

      // Store challenge in session for later verification (expires in 5 minutes)
      req.session.passkeyChallenge = options.challenge;
      req.session.passkeyChallengeExpiry = Date.now() + 5 * 60 * 1000;

      return res.json(options);
    } catch (error) {
      console.error('Passkey registration init failed:', error);
      next(error); // Pass to global error handler
    }
  });

  // 2. Verify passkey registration response from client
  app.post('/auth/passkey/register/verify', async (req: Request, res: Response, next: NextFunction) => {
    try {
      const { userId } = req.session;
      const { passkeyChallenge, passkeyChallengeExpiry } = req.session;

      // Validate session and challenge
      if (!userId) {
        return res.status(401).json({ error: 'Unauthenticated' });
      }
      if (!passkeyChallenge || Date.now() > passkeyChallengeExpiry) {
        return res.status(400).json({ error: 'Invalid or expired challenge' });
      }

      const user = await UserService.findById(userId);
      if (!user) {
        return res.status(404).json({ error: 'User not found' });
      }

      // Verify the registration response from the client
      const verification = await verifyRegistrationResponse({
        response: req.body,
        expectedChallenge: passkeyChallenge,
        expectedOrigin: origin,
        expectedRPID: rpID,
        requireUserVerification: true,
      });

      if (!verification.verified) {
        return res.status(400).json({ error: 'Passkey verification failed' });
      }

      // Store the new authenticator for the user
      const { authenticatorInfo } = verification;
      await UserService.addAuthenticator(userId, {
        id: authenticatorInfo?.credentialID || uuidv4(),
        publicKey: authenticatorInfo?.credentialPublicKey,
        counter: authenticatorInfo?.counter || 0,
        transports: req.body.transports || [],
      });

      // Clear challenge from session
      delete req.session.passkeyChallenge;
      delete req.session.passkeyChallengeExpiry;

      return res.json({ verified: true, userId });
    } catch (error) {
      console.error('Passkey registration verify failed:', error);
      next(error);
    }
  });
}

# Phishing Simulation Tester: Measures auth phishing susceptibility
# Dependencies: pip install selenium webdriver-manager pytest
from selenium import webdriver
from selenium.webdriver.chrome.service import Service as ChromeService
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from webdriver_manager.chrome import ChromeDriverManager
import pytest
import time
import json

class PhishingSimulator:
    def __init__(self, target_login_url: str, phishing_page_path: str, headless: bool = True):
        self.target_login_url = target_login_url
        self.phishing_page_path = phishing_page_path
        self.headless = headless
        self.results = {
            "total_attempts": 0,
            "successful_phishes": 0,
            "failed_attempts": 0,
            "credential_handovers": 0
        }

    def _init_driver(self) -> webdriver.Chrome:
        """Initialize headless Chrome driver with security settings disabled for testing"""
        chrome_options = Options()
        if self.headless:
            chrome_options.add_argument("--headless=new")
        chrome_options.add_argument("--no-sandbox")
        chrome_options.add_argument("--disable-dev-shm-usage")
        # Disable phishing warnings for testing (DON'T USE IN PRODUCTION)
        chrome_options.add_argument("--disable-features=PhishingDetector")
        chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
        chrome_options.add_experimental_option('useAutomationExtension', False)

        driver = webdriver.Chrome(
            service=ChromeService(ChromeDriverManager().install()),
            options=chrome_options
        )
        return driver

    def run_simulation(self, test_credentials: dict, num_attempts: int = 100) -> dict:
        """Run N simulated phishing attempts with test credentials"""
        for attempt in range(num_attempts):
            driver = None
            try:
                driver = self._init_driver()
                self.results["total_attempts"] += 1

                # Step 1: Navigate to phishing page
                driver.get(f"file://{self.phishing_page_path}")
                WebDriverWait(driver, 10).until(
                    EC.presence_of_element_located((By.ID, "phishing-email"))
                )

                # Step 2: Enter test credentials (simulate user falling for phish)
                driver.find_element(By.ID, "phishing-email").send_keys(test_credentials["email"])
                driver.find_element(By.ID, "phishing-password").send_keys(test_credentials["password"])
                driver.find_element(By.ID, "phishing-submit").click()

                # Step 3: Check if credentials were "exfiltrated" (logged to simulation server)
                WebDriverWait(driver, 10).until(
                    EC.presence_of_element_located((By.ID, "exfil-success"))
                )
                self.results["successful_phishes"] += 1
                self.results["credential_handovers"] += 1

                # Simulate MFA step if target uses MFA
                if self._is_mfa_required(driver):
                    mfa_code = "123456" # Test MFA code
                    driver.find_element(By.ID, "mfa-input").send_keys(mfa_code)
                    driver.find_element(By.ID, "mfa-submit").click()
                    self.results["credential_handovers"] += 1

            except Exception as e:
                self.results["failed_attempts"] += 1
                print(f"Attempt {attempt} failed: {str(e)}")
            finally:
                if driver:
                    driver.quit()
                time.sleep(0.5) # Rate limit to avoid detection

        # Calculate success rate
        self.results["success_rate"] = (
            self.results["successful_phishes"] / self.results["total_attempts"]
        ) * 100 if self.results["total_attempts"] > 0 else 0
        return self.results

    def _is_mfa_required(self, driver: webdriver.Chrome) -> bool:
        """Check if the phishing flow triggered an MFA prompt"""
        try:
            driver.find_element(By.ID, "mfa-input")
            return True
        except:
            return False

    def save_results(self, output_path: str) -> None:
        """Save simulation results to JSON"""
        with open(output_path, "w") as f:
            json.dump(self.results, f, indent=2)

# Example usage (run with pytest)
@pytest.mark.parametrize("num_attempts", [100])
def test_phishing_susceptibility(num_attempts):
    simulator = PhishingSimulator(
        target_login_url="https://auth.acme.example/login",
        phishing_page_path="./test-phishing-page.html",
        headless=True
    )
    results = simulator.run_simulation(
        test_credentials={"email": "test@example.com", "password": "test-password"},
        num_attempts=num_attempts
    )
    simulator.save_results("./phishing-sim-results.json")
    print(f"Phishing success rate: {results['success_rate']}%")
    assert results["success_rate"] < 5.0 # Fail if success rate is above 5%

// Real-Time Phishing Auth Monitor with Prometheus Metrics
// Dependencies: go get github.com/prometheus/client_golang/prometheus github.com/prometheus/client_golang/prometheus/promhttp
package main

import (
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "sync"
    "time"

    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promhttp"
)

// AuthAttempt represents a single authentication attempt
type AuthAttempt struct {
    Timestamp    time.Time `json:"timestamp"`
    UserID       string    `json:"user_id"`
    IPAddress    string    `json:"ip_address"`
    UserAgent    string    `json:"user_agent"`
    AuthMethod   string    `json:"auth_method"` // password, sms_otp, passkey, etc.
    Success      bool      `json:"success"`
    IsPhishing   bool      `json:"is_phishing"` // Flagged by rules engine
}

// PhishingMonitor tracks auth attempts and exposes Prometheus metrics
type PhishingMonitor struct {
    mu              sync.RWMutex
    attempts        []AuthAttempt
    phishingRules   []PhishingRule
    metrics         *MonitorMetrics
}

// MonitorMetrics holds Prometheus metrics for auth monitoring
type MonitorMetrics struct {
    totalAttempts   prometheus.Counter
    phishingAttempts prometheus.Counter
    authSuccess     *prometheus.CounterVec
    authFailure     *prometheus.CounterVec
}

// PhishingRule defines a rule to flag suspicious attempts
type PhishingRule interface {
    Evaluate(attempt AuthAttempt, history []AuthAttempt) bool
}

// IPFrequencyRule flags IPs with >10 failed attempts in 5 minutes
type IPFrequencyRule struct {
    MaxAttempts int
    Window      time.Duration
}

func (r IPFrequencyRule) Evaluate(attempt AuthAttempt, history []AuthAttempt) bool {
    if attempt.Success {
        return false
    }
    cutoff := time.Now().Add(-r.Window)
    count := 0
    for _, hist := range history {
        if hist.IPAddress == attempt.IPAddress && hist.Timestamp.After(cutoff) && !hist.Success {
            count++
        }
    }
    return count > r.MaxAttempts
}

// ImpossibleTravelRule flags attempts from geographically distant IPs in <1 hour
type ImpossibleTravelRule struct {
    MaxDistanceKm float64
    Window        time.Duration
}

func (r ImpossibleTravelRule) Evaluate(attempt AuthAttempt, history []AuthAttempt) bool {
    // Simplified: Assume we have GeoIP data; here we mock distance calculation
    return false // Placeholder for actual GeoIP logic
}

func NewPhishingMonitor() *PhishingMonitor {
    metrics := &MonitorMetrics{
        totalAttempts: prometheus.NewCounter(prometheus.CounterOpts{
            Name: "auth_total_attempts",
            Help: "Total number of authentication attempts",
        }),
        phishingAttempts: prometheus.NewCounter(prometheus.CounterOpts{
            Name: "auth_phishing_attempts",
            Help: "Total number of flagged phishing attempts",
        }),
        authSuccess: prometheus.NewCounterVec(prometheus.CounterOpts{
            Name: "auth_success_total",
            Help: "Total successful auth attempts by method",
        }, []string{"method"}),
        authFailure: prometheus.NewCounterVec(prometheus.CounterOpts{
            Name: "auth_failure_total",
            Help: "Total failed auth attempts by method",
        }, []string{"method"}),
    }

    // Register metrics with Prometheus
    prometheus.MustRegister(metrics.totalAttempts)
    prometheus.MustRegister(metrics.phishingAttempts)
    prometheus.MustRegister(metrics.authSuccess)
    prometheus.MustRegister(metrics.authFailure)

    return &PhishingMonitor{
        attempts: make([]AuthAttempt, 0),
        phishingRules: []PhishingRule{
            IPFrequencyRule{MaxAttempts: 10, Window: 5 * time.Minute},
            ImpossibleTravelRule{MaxDistanceKm: 500, Window: 1 * time.Hour},
        },
        metrics: metrics,
    }
}

// RecordAttempt records a new auth attempt and checks for phishing
func (pm *PhishingMonitor) RecordAttempt(attempt AuthAttempt) {
    pm.mu.Lock()
    defer pm.mu.Unlock()

    // Check all phishing rules
    isPhishing := false
    for _, rule := range pm.phishingRules {
        if rule.Evaluate(attempt, pm.attempts) {
            isPhishing = true
            break
        }
    }
    attempt.IsPhishing = isPhishing
    pm.attempts = append(pm.attempts, attempt)

    // Update metrics
    pm.metrics.totalAttempts.Inc()
    if isPhishing {
        pm.metrics.phishingAttempts.Inc()
    }
    if attempt.Success {
        pm.metrics.authSuccess.WithLabelValues(attempt.AuthMethod).Inc()
    } else {
        pm.metrics.authFailure.WithLabelValues(attempt.AuthMethod).Inc()
    }

    // Trim attempts older than 24 hours to save memory
    cutoff := time.Now().Add(-24 * time.Hour)
    newAttempts := make([]AuthAttempt, 0)
    for _, a := range pm.attempts {
        if a.Timestamp.After(cutoff) {
            newAttempts = append(newAttempts, a)
        }
    }
    pm.attempts = newAttempts
}

// HTTP handler to receive auth attempt webhooks
func (pm *PhishingMonitor) handleAuthWebhook(w http.ResponseWriter, r *http.Request) {
    if r.Method != http.MethodPost {
        http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
        return
    }

    var attempt AuthAttempt
    if err := json.NewDecoder(r.Body).Decode(&attempt); err != nil {
        http.Error(w, "Invalid request body", http.StatusBadRequest)
        return
    }
    attempt.Timestamp = time.Now()

    pm.RecordAttempt(attempt)
    w.WriteHeader(http.StatusAccepted)
}

func main() {
    monitor := NewPhishingMonitor()

    http.HandleFunc("/webhook/auth", monitor.handleAuthWebhook)
    http.Handle("/metrics", promhttp.Handler())

    log.Println("Phishing auth monitor running on :8080")
    log.Fatal(http.ListenAndServe(":8080", nil))
}

# Quick MFA fatigue test snippet
simulator = PhishingSimulator(
    target_login_url="https://your-app.com/login",
    phishing_page_path="./mfa-fatigue-phish.html",
    headless=True
)
results = simulator.run_simulation(
    test_credentials={"email": "test@your-app.com", "password": "test"},
    num_attempts=50
)
print(f"MFA fatigue success rate: {results['success_rate']}%")

// Add authenticator ID to JWT session
import jwt from 'jsonwebtoken';

function createSessionToken(userId: string, authenticatorId: string) {
  return jwt.sign(
    {
      sub: userId,
      authMethod: 'passkey',
      authenticatorId: authenticatorId, // Bind session to this passkey
      exp: Math.floor(Date.now() / 1000) + 60 * 60 * 8 // 8 hour expiry
    },
    process.env.JWT_SECRET!
  );
}

# Prometheus alert rule for phishing attempts
groups:
- name: auth-alerts
  rules:
  - alert: HighPhishingAttempts
    expr: rate(auth_phishing_attempts[5m]) > 0.5
    for: 1m
    labels:
      severity: critical
    annotations:
      summary: "High phishing attempt rate detected"
      description: "{{ $value }} phishing attempts per second in the last 5 minutes"