# ═══════════════════════════════════════════════════════════════════════════════ # File: app/core/engagement_delay.py # Description: Engagement Time Maximization (Real Honeypot Feature) # ═══════════════════════════════════════════════════════════════════════════════ """ Engagement Time Delay System. Real honeypots aim to waste attacker time. This module adds realistic delays to maximize engagement. Reference: "Wasting Scammer Time" - USENIX Security 2022 """ import asyncio import random from typing import Optional, Dict from enum import Enum from app.config import settings from app.utils.logger import AgentLogger class TypingDelaySimulator: """ Calculate realistic delay times for different persona types. """ SPEED_PROFILES = { "very_slow": {"base_delay": 45, "per_char": 0.5}, # Elderly "slow": {"base_delay": 20, "per_char": 0.3}, # Middle-aged/stressed "medium": {"base_delay": 10, "per_char": 0.1}, # Normal adult "fast": {"base_delay": 5, "per_char": 0.05} # Youth/tech-savvy } @classmethod def get_delay_seconds(cls, text: str, typing_speed: str = "medium", stress_level: str = "normal") -> int: """Calculate realistic typing delay in seconds.""" profile = cls.SPEED_PROFILES.get(typing_speed, cls.SPEED_PROFILES["medium"]) base = profile["base_delay"] per_char = profile["per_char"] # Stress increases delay (distracted/panicking) stress_multiplier = 1.0 if stress_level == "high": stress_multiplier = 1.5 elif stress_level == "low": stress_multiplier = 0.8 delay = int((base + len(text) * per_char) * stress_multiplier) # Add randomness (humans aren't consistent) delay += random.randint(-5, 10) if settings.DEBUG or not settings.ENABLE_ENGAGEMENT_DELAY: return 0 # No delay in debug/hackathon mode return max(2, delay) # Minimum 2 seconds @classmethod def get_delay_excuse(cls, delay_seconds: int, language: str = "hinglish") -> Optional[str]: """Generate a natural excuse for the delay.""" if delay_seconds < 20: return None # No excuse needed for quick replies excuses = { "hinglish": [ f"sorry, {delay_seconds//10} min lage... phone down tha", "ruko typing kar rahi thi...", "arre connection weak tha...", "message type karte karte delete ho gaya tha" ] } return random.choice(excuses.get(language, excuses["hinglish"])) class DelayType(str, Enum): """Types of delays to simulate.""" TYPING = "typing" # Simulating typing THINKING = "thinking" # Pretending to think BANK_ERROR = "bank_error" # Fake bank server issues OTP_WAIT = "otp_wait" # Waiting for OTP NETWORK = "network" # Network lag simulation class EngagementDelayer: """ Adds realistic delays to maximize scammer engagement time. Delays are randomized to appear natural. """ # Delay ranges in seconds (min, max) DELAY_CONFIGS = { DelayType.TYPING: (1.0, 3.0), # Faster typing for API response DelayType.THINKING: (1.0, 3.5), # Reduced thinking time DelayType.BANK_ERROR: (2.0, 4.0), # Capped at 4s DelayType.OTP_WAIT: (2.0, 4.0), # Capped at 4s for API stability DelayType.NETWORK: (0.1, 1.0), # Fast network } # Messages to display during delay (for personas) DELAY_MESSAGES = { DelayType.TYPING: [ "Let me type that for you...", "One moment, typing slowly...", "My eyes are not so good, typing...", ], DelayType.THINKING: [ "Let me think about this...", "Hmm, I need to consider...", "Wait, I'm confused...", ], DelayType.BANK_ERROR: [ "The bank app is loading slowly...", "Server error, trying again...", "Internet is slow today...", ], DelayType.OTP_WAIT: [ "Waiting for OTP... not received yet...", "OTP is delayed, checking again...", "Still waiting for the message...", ], DelayType.NETWORK: [ "Connection is slow...", "Loading...", ] } def __init__(self, enabled: bool = True): self.enabled = settings.ENABLE_ENGAGEMENT_DELAY self.total_delay_seconds = 0.0 self.logger = AgentLogger("engagement_delayer") async def delay( self, delay_type: DelayType = DelayType.TYPING, custom_min: Optional[float] = None, custom_max: Optional[float] = None ) -> float: """ Apply an async delay. Args: delay_type: Type of delay to simulate custom_min: Optional custom minimum seconds custom_max: Optional custom maximum seconds Returns: Actual seconds delayed """ if not self.enabled: return 0.0 min_delay, max_delay = self.DELAY_CONFIGS.get( delay_type, (1.0, 3.0) ) if custom_min is not None: min_delay = custom_min if custom_max is not None: max_delay = custom_max delay_seconds = random.uniform(min_delay, max_delay) await asyncio.sleep(delay_seconds) self.total_delay_seconds += delay_seconds return delay_seconds def get_delay_message(self, delay_type: DelayType) -> str: """Get a random delay message for the given type.""" messages = self.DELAY_MESSAGES.get(delay_type, ["Please wait..."]) return random.choice(messages) async def simulate_typing(self, message_length: int = 50) -> float: """ Simulate realistic typing delay based on message length. Average typing speed: 40 WPM = ~200 CPM = 3.3 chars/sec Elderly personas type slower: ~1.5 chars/sec """ if not self.enabled: return 0.0 # Assume slow typing (elderly persona) chars_per_second = random.uniform(1.0, 2.5) delay = message_length / chars_per_second # Cap at reasonable max delay = min(delay, 4.0) # Cap for API stability await asyncio.sleep(delay) self.total_delay_seconds += delay return delay async def simulate_bank_issue(self) -> tuple[float, str]: """ Simulate a fake bank server issue for time wasting. Returns: Tuple of (delay_seconds, excuse_message) """ delay = await self.delay(DelayType.BANK_ERROR) excuses = [ "Arey, bank app showing error. Let me try again...", "Server not responding. Internet is slow today...", "Transaction failed. Waiting for it to process...", "App is hanging. My phone is old, please wait...", "Getting 'technical error'. What should I do?", ] return delay, random.choice(excuses) async def simulate_otp_delay(self) -> tuple[float, str]: """ Simulate waiting for OTP (great time waster). Returns: Tuple of (delay_seconds, status_message) """ delay = await self.delay(DelayType.OTP_WAIT) statuses = [ "OTP not received yet... checking phone...", "Still waiting... network is slow maybe...", "No message yet. Should I request again?", "Got a different OTP earlier. This one is new?", ] return delay, random.choice(statuses) def get_total_time_wasted(self) -> float: """Get total time wasted on scammer in seconds.""" return self.total_delay_seconds def get_stats(self) -> dict: """Get engagement delay statistics.""" return { "total_delay_seconds": round(self.total_delay_seconds, 2), "total_delay_formatted": f"{int(self.total_delay_seconds // 60)}m {int(self.total_delay_seconds % 60)}s", "enabled": self.enabled } # Global instance engagement_delayer = EngagementDelayer(enabled=True) __all__ = ["EngagementDelayer", "DelayType", "engagement_delayer"]