Endpoint Detection and Response (EDR): Complete Deployment and Threat Detection Guide

from dataclasses import dataclass
from typing import List, Dict, Optional
from enum import Enum


class OSType(Enum):
    WINDOWS = "windows"
    LINUX = "linux"
    MACOS = "macos"
    ANDROID = "android"
    IOS = "ios"


@dataclass
class Endpoint:
    """Represents an endpoint to be protected."""
    hostname: str
    ip_address: str
    os_type: OSType
    os_version: str
    department: str
    criticality: str  # critical, high, medium, low
    owner: str


class EDRDeployment:
    """Manage EDR deployment across organization."""
    
    def __init__(self, edr_vendor: str, tenant_id: str):
        self.edr_vendor = edr_vendor
        self.tenant_id = tenant_id
        self.deployed_agents = {}
        self.pending_deployments = []
    
    def plan_deployment(self, endpoints: List[Endpoint]) -> Dict:
        """Plan deployment phases based on criticality."""
        
        # Group by criticality
        by_criticality = {
            'critical': [],
            'high': [],
            'medium': [],
            'low': []
        }
        
        for endpoint in endpoints:
            by_criticality[endpoint.criticality].append(endpoint)
        
        # Create deployment phases
        phases = [
            {
                'phase': 1,
                'name': 'Critical Assets',
                'endpoints': len(by_criticality['critical']),
                'description': 'Domain controllers, critical servers',
                'duration_days': 7
            },
            {
                'phase': 2,
                'name': 'High Priority',
                'endpoints': len(by_criticality['high']),
                'description': 'Executive machines, sensitive data servers',
                'duration_days': 14
            },
            {
                'phase': 3,
                'name': 'Standard Deployment',
                'endpoints': len(by_criticality['medium']),
                'description': 'General user workstations',
                'duration_days': 30
            },
            {
                'phase': 4,
                'name': 'Remaining Endpoints',
                'endpoints': len(by_criticality['low']),
                'description': 'Low priority and test devices',
                'duration_days': 14
            }
        ]
        
        return {
            'total_endpoints': len(endpoints),
            'phases': phases,
            'estimated_completion_days': sum(p['duration_days'] for p in phases)
        }
    
    def deploy_agent(self, endpoint: Endpoint) -> Dict:
        """Deploy EDR agent to an endpoint."""
        
        # Simulate agent deployment
        agent_id = f"{self.edr_vendor}-{endpoint.hostname}-{endpoint.os_type.value}"
        
        self.deployed_agents[agent_id] = {
            'endpoint': endpoint,
            'status': 'active',
            'version': '7.15.0',
            'deployed_at': '2026-03-13T10:00:00Z',
            'last_heartbeat': '2026-03-13T10:05:00Z'
        }
        
        return {
            'agent_id': agent_id,
            'status': 'deployed',
            'endpoint': endpoint.hostname,
            'os': endpoint.os_type.value
        }
    
    def deploy_to_group(self, endpoints: List[Endpoint]) -> List[Dict]:
        """Deploy agents to a group of endpoints."""
        
        results = []
        for endpoint in endpoints:
            result = self.deploy_agent(endpoint)
            results.append(result)
        
        return results

from dataclasses import dataclass
from typing import Dict, List


@dataclass
class EDRConfig:
    """EDR agent configuration."""
    
    # Real-time protection
    real_time_protection: bool = True
    scan_on_access: bool = True
    
    # Behavioral monitoring
    process_monitoring: bool = True
    file_monitoring: bool = True
    registry_monitoring: bool = True
    network_monitoring: bool = True
    
    # Cloud connectivity
    cloud_upload: bool = True
    upload_suspicious: bool = True
    
    # Isolation
    allow_isolation: bool = True
    
    # Exclusions (be careful!)
    exclusions: List[str] = None
    
    def to_dict(self) -> Dict:
        """Convert to configuration dictionary."""
        return {
            'realTimeProtection': self.real_time_protection,
            'scanOnAccess': self.scan_on_access,
            'processMonitoring': self.process_monitoring,
            'fileMonitoring': self.file_monitoring,
            'registryMonitoring': self.registry_monitoring,
            'networkMonitoring': self.network_monitoring,
            'cloudUpload': self.cloud_upload,
            'uploadSuspiciousFiles': self.upload_suspicious,
            'allowIsolation': self.allow_isolation,
            'exclusions': self.exclusions or []
        }


# Production configuration
production_config = EDRConfig(
    real_time_protection=True,
    scan_on_access=True,
    process_monitoring=True,
    file_monitoring=True,
    registry_monitoring=True,
    network_monitoring=True,
    cloud_upload=True,
    upload_suspicious=True,
    allow_isolation=True,
    exclusions=[
        r'C:\Windows\System32\',
        r'C:\Windows\SysWOW64\',
        r'C:\Program Files\Antivirus\',
    ]
)

import hashlib
from datetime import datetime
from typing import Dict, List, Optional


class ThreatDetector:
    """Detect threats using multiple techniques."""
    
    def __init__(self):
        self.signatures = {}
        self.behavioral_rules = []
        self.ml_models = []
        self.threat_intel = {}
    
    def add_signature(self, name: str, indicator: str, indicator_type: str):
        """Add detection signature."""
        self.signatures[name] = {
            'indicator': indicator,
            'type': indicator_type,
            'name': name
        }
    
    def detect_signature(self, event: Dict) -> Optional[Dict]:
        """Detect based on known signatures."""
        
        for name, sig in self.signatures.items():
            if sig['type'] == 'hash':
                if event.get('file_hash') == sig['indicator']:
                    return {
                        'detection': 'signature',
                        'rule': name,
                        'severity': 'high',
                        'indicator': sig['indicator']
                    }
            elif sig['type'] == 'process':
                if sig['indicator'].lower() in event.get('command_line', '').lower():
                    return {
                        'detection': 'signature',
                        'rule': name,
                        'severity': 'high',
                        'indicator': sig['indicator']
                    }
            elif sig['type'] == 'network':
                if sig['indicator'] in event.get('remote_ip', ''):
                    return {
                        'detection': 'signature',
                        'rule': name,
                        'severity': 'high',
                        'indicator': sig['indicator']
                    }
        
        return None
    
    def detect_behavioral(self, event: Dict, context: Dict) -> Optional[Dict]:
        """Detect based on behavioral rules."""
        
        rules = [
            {
                'name': 'Suspicious PowerShell',
                'condition': lambda e: 'powershell' in e.get('process', '').lower() 
                    and ('-nop' in e.get('command_line', '') 
                         or '-enc' in e.get('command_line', '')),
                'severity': 'high'
            },
            {
                'name': 'Lsass Access',
                'condition': lambda e: 'lsass.exe' in e.get('target_process', '').lower(),
                'severity': 'critical'
            },
            {
                'name': 'Suspicious Scheduled Task',
                'condition': lambda e: 'schtasks' in e.get('command_line', '').lower()
                    and '/create' in e.get('command_line', '').lower(),
                'severity': 'medium'
            },
            {
                'name': 'Registry Modification',
                'condition': lambda e: e.get('operation') == 'registry_modify'
                    and 'run' in e.get('key', '').lower(),
                'severity': 'high'
            },
            {
                'name': 'Mass File Access',
                'condition': lambda e: e.get('operation') == 'file_access'
                    and e.get('file_count', 0) > 1000,
                'severity': 'medium'
            }
        ]
        
        for rule in rules:
            if rule['condition'](event):
                return {
                    'detection': 'behavioral',
                    'rule': rule['name'],
                    'severity': rule['severity'],
                    'event': event
                }
        
        return None
    
    def analyze_file(self, file_path: str) -> Dict:
        """Analyze file for malicious indicators."""
        
        # Calculate hash
        with open(file_path, 'rb') as f:
            content = f.read()
            sha256 = hashlib.sha256(content).hexdigest()
        
        # Check against threat intel
        if sha256 in self.threat_intel:
            return {
                'malicious': True,
                'threat': self.threat_intel[sha256],
                'sha256': sha256
            }
        
        # Check file characteristics
        characteristics = {
            'is_pe': content[:2] == b'MZ',
            'is_dll': b'DLL' in content[:1000],
            'packed': self._check_packing(content),
            'suspicious_imports': self._check_imports(content)
        }
        
        return {
            'malicious': False,
            'characteristics': characteristics,
            'sha256': sha256
        }

# Example detection rule in Sigma format
DETECTION_RULE = """
title: Suspicious PowerShell Execution
id: 31112d87-7e9f-4f1c-a1c8-5c1c8c9c9c9c
status: stable
description: Detects suspicious PowerShell execution with encoded commands
author: Security Team
date: 2026/03/13
tags:
    - attack.execution
    - attack.t1059.001
    - attack.defense_evasion
    - attack.t1027

logsource:
    category: process_creation
    product: windows
    service: security

detection:
    selection:
        Image|endswith: '\powershell.exe'
        CommandLine|contains:
            - '-nop'
            - '-w hidden'
            - '-enc '
            - 'bypass'
            - 'DownloadString'
            - 'DownloadFile'
            - 'IEX '
    condition: selection
level: high
"""

# YARA rule example
YARA_RULE = """
rule Suspicious_Dropper
{
    meta:
        author = "Security Team"
        description = "Detects suspicious file dropper"
        date = "2026-03-13"
    
    strings:
        $s1 = "CreateFile" nocase
        $s2 = "WriteFile" nocase
        $s3 = "GetTempPath" nocase
        $s4 = ".exe" nocase
        
    condition:
        3 of them
}
"""

from datetime import datetime
from enum import Enum
from typing import List, Dict


class Severity(Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
    INFO = "info"


class IncidentStatus(Enum):
    OPEN = "open"
    INVESTIGATING = "investigating"
    CONTAINED = "contained"
    ERADICATED = "eradicated"
    RECOVERED = "recovered"
    CLOSED = "closed"


class Incident:
    """Security incident with EDR integration."""
    
    def __init__(self, title: str, severity: Severity):
        self.id = f"INC-{datetime.now().strftime('%Y%m%d')}-{hash(title) % 10000}"
        self.title = title
        self.severity = severity
        self.status = IncidentStatus.OPEN
        self.detected_at = datetime.now()
        self.events = []
        self.affected_endpoints = []
        self.response_actions = []
    
    def add_event(self, event: Dict):
        """Add related event to incident."""
        self.events.append(event)
    
    def add_endpoint(self, endpoint_id: str):
        """Add affected endpoint."""
        if endpoint_id not in self.affected_endpoints:
            self.affected_endpoints.append(endpoint_id)
    
    def respond(self, action: str, target: str, reason: str):
        """Record response action."""
        self.response_actions.append({
            'action': action,
            'target': target,
            'reason': reason,
            'timestamp': datetime.now(),
            'performed_by': 'Security Analyst'
        })
    
    def isolate_endpoint(self, endpoint_id: str, edr_client):
        """Isolate endpoint from network."""
        
        # Issue isolation command via EDR
        result = edr_client.isolate_endpoint(endpoint_id)
        
        self.respond('isolate', endpoint_id, 'Malware detected')
        
        return {
            'status': 'success' if result else 'failed',
            'endpoint': endpoint_id,
            'isolated': result
        }
    
    def kill_process(self, endpoint_id: str, process_id: int, edr_client):
        """Kill malicious process."""
        
        result = edr_client.kill_process(endpoint_id, process_id)
        
        self.respond('kill_process', f"{endpoint_id}:{process_id}", 'Malicious process')
        
        return result
    
    def collect_forensics(self, endpoint_id: str, edr_client, artifacts: List[str]):
        """Collect forensic artifacts."""
        
        collection = {}
        
        for artifact in artifacts:
            data = edr_client.collect_artifact(endpoint_id, artifact)
            collection[artifact] = data
        
        self.respond('forensics_collect', endpoint_id, f"Collected {len(artifacts)} artifacts")
        
        return collection
    
    def remediate(self, endpoint_id: str, action: str):
        """Remediate endpoint."""
        
        self.respond('remediate', endpoint_id, action)
        self.status = IncidentStatus.ERADICATED


# Automated response playbook
AUTOMATED_PLAYBOOK = {
    'critical_malware': [
        {'action': 'isolate', 'target': 'endpoint'},
        {'action': 'collect_forensics', 'target': 'memory_dump, disk'},
        {'action': 'kill_process', 'target': 'malware_process'},
        {'action': 'quarantine_file', 'target': 'malware_file'}
    ],
    'suspicious_process': [
        {'action': 'alert', 'target': 'security_team'},
        {'action': 'increase_monitoring', 'target': 'endpoint'}
    ],
    'data_exfiltration': [
        {'action': 'block_network', 'target': 'endpoint'},
        {'action': 'collect_forensics', 'target': 'network_logs'}
    ]
}

from typing import List, Dict, Callable
import itertools


class ThreatHunt:
    """Proactive threat hunting framework."""
    
    def __init__(self, edr_client):
        self.edr = edr_client
        self.hypotheses = []
        self.findings = []
    
    def add_hypothesis(self, name: str, description: str, query: str):
        """Add hunting hypothesis."""
        self.hypotheses.append({
            'name': name,
            'description': description,
            'query': query,
            'status': 'pending'
        })
    
    def run_hypothesis(self, hypothesis: Dict) -> List[Dict]:
        """Run a hunting hypothesis."""
        
        # Execute query
        results = self.edr.query(hypothesis['query'])
        
        if results:
            hypothesis['status'] = 'found'
            hypothesis['results'] = results
            self.findings.extend(results)
        else:
            hypothesis['status'] = 'not_found'
        
        return results
    
    def run_all(self):
        """Run all hypotheses."""
        for hypothesis in self.hypotheses:
            self.run_hypothesis(hypothesis)
        
        return {
            'hypotheses_run': len(self.hypotheses),
            'findings': len(self.findings)
        }


# Example hunting queries
HUNTING_QUERIES = [
    {
        'name': 'Ransomware Indicators',
        'description': 'Look for ransomware encryption activity',
        'query': '''
            process_events 
            | where file_extension in~ (".encrypted", ".locked", ".crypto")
            | where operation == "file_created"
            | where file_count > 100
        '''
    },
    {
        'name': 'Lateral Movement',
        'description': 'Detect lateral movement via RDP',
        'query': '''
            network_events
            | where destination_port == 3389
            | where connection_count > 10
        '''
    },
    {
        'name': 'Privilege Escalation',
        'description': 'Detect privilege escalation attempts',
        'query': '''
            process_events
            | where command_line contains "SeDebugPrivilege"
               or command_line contains "SeTakeOwnershipPrivilege"
        '''
    },
    {
        'name': 'Suspicious Scheduled Tasks',
        'description': 'Look for malicious scheduled tasks',
        'query': '''
            scheduled_task_events
            | where action contains "powershell"
               or action contains "cmd.exe"
        '''
    },
    {
        'name': 'Beaconing Detection',
        'description': 'Detect C2 beaconing patterns',
        'query': '''
            network_events
            | summarize connections = count(), 
                        intervals = make_list(timestamp) 
            by remote_ip, process_name
            | where connections > 100
            | where avg_interval between (55min .. 65min)
        '''
    }
]

class EDRSIEMIntegration:
    """Integrate EDR with SIEM."""
    
    def __init__(self, edr_client, siem_client):
        self.edr = edr_client
        self.siem = siem_client
    
    def send_alerts_to_siem(self):
        """Forward EDR alerts to SIEM."""
        
        # Get recent alerts
        alerts = self.edr.get_alerts(status='open')
        
        for alert in alerts:
            # Transform to SIEM format
            siem_event = self.transform_alert(alert)
            
            # Send to SIEM
            self.siem.send_event(siem_event)
    
    def transform_alert(self, alert: Dict) -> Dict:
        """Transform EDR alert to SIEM format."""
        
        return {
            'timestamp': alert['timestamp'],
            'source': 'edr',
            'vendor': self.edr.vendor_name,
            'rule_name': alert['rule_name'],
            'severity': alert['severity'],
            'endpoint': alert['endpoint_hostname'],
            'process': alert.get('process_name'),
            'command_line': alert.get('command_line'),
            'file_hash': alert.get('file_hash'),
            'network_connection': alert.get('remote_ip')
        }
    
    def enrich_with_threat_intel(self, indicator: str) -> Dict:
        """Enrich indicator with threat intelligence."""
        
        # Query threat intel sources
        intel = self.siem.query_threat_intel(indicator)
        
        return {
            'indicator': indicator,
            'known_malicious': intel.get('malicious', False),
            'threat_actors': intel.get('threat_actors', []),
            'campaigns': intel.get('campaigns', []),
            'first_seen': intel.get('first_seen'),
            'tags': intel.get('tags', [])
        }