import win32evtlog import json import requests import os import sys import logging from datetime import datetime # File to store the last processed record number LAST_PROCESSED_FILE = 'last_processed_record.txt' # CrowdStrike API credentials client_id = 'xxxxxx' client_secret = 'yyyyy' base_url = "https://api.us-2.crowdstrike.com" # Configuration server = 'localhost' log_name = 'Data Security Essentials' source_names = ['Superna Data Security Essentials BOT Service', 'Superna Data Security Policy Engine'] trigger_severities = ['MAJOR', 'CRITICAL'] # Configure severities that should trigger containment enable_isolation = True # Set to False to disable isolation actions def setup_logging(): """Set up logging to file and console.""" script_directory = os.path.dirname(os.path.abspath(__file__)) log_file_path = os.path.join(script_directory, "crowdstrike-dse.log") logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(log_file_path, mode='w'), logging.StreamHandler(sys.stdout) ]) logging.info(f"Script run on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") logging.info(f"Logging to file: {log_file_path}") def get_last_processed_record(): """Retrieve the last processed event record number and initialize if the file doesn't exist.""" if os.path.exists(LAST_PROCESSED_FILE): try: with open(LAST_PROCESSED_FILE, 'r') as f: lines = f.readlines() if len(lines) >= 2: last_record = int(lines[0].strip()) last_time_and_event_id = lines[1].strip() logging.info(f"Loaded last processed record: {last_record}, {last_time_and_event_id}") return last_record, last_time_and_event_id except Exception as e: logging.error(f"Error reading {LAST_PROCESSED_FILE}: {e}") logging.info(f"No last processed record file found. Starting fresh.") return 0, None def save_last_processed_record(record_number, event_time, event_id): """Save the last processed event record number, time, and EventID to a file.""" try: with open(LAST_PROCESSED_FILE, 'w') as f: f.write(f"{record_number}\n") f.write(f"{event_time} | EventID: {event_id}\n") logging.info(f"Saved last processed record: {record_number}, {event_time}, EventID: {event_id}") except Exception as e: logging.error(f"Error saving {LAST_PROCESSED_FILE}: {e}") def collect_windows_event_log(server, log_name, source_names): """Collect event logs from the specified Windows Event Log and extract relevant client IPs.""" logging.info(f"Collecting Windows Event Logs from {log_name} on server {server}...") h = win32evtlog.OpenEventLog(server, log_name) flags = win32evtlog.EVENTLOG_BACKWARDS_READ | win32evtlog.EVENTLOG_SEQUENTIAL_READ last_processed_record, last_time_and_event_id = get_last_processed_record() logs = [] latest_record = last_processed_record latest_event_time = None latest_event_id = None try: while True: events = win32evtlog.ReadEventLog(h, flags, 0) if not events: logging.debug("No more events to read.") break for event in events: record_number = event.RecordNumber event_time = event.TimeGenerated.Format() event_id = event.EventID & 0xFFFF # Skip already processed events if record_number <= last_processed_record: continue # Filter by source names if event.SourceName not in source_names: continue # Log event metadata logging.info(f"Processing Event - RecordNumber: {record_number}, EventID: {event_id}, TimeGenerated: {event_time}, Source: {event.SourceName}") try: # Extract and log JSON payload from Data if event.Data: data_content = event.Data.decode('utf-8', errors='replace') logging.debug(f"Raw JSON Payload: {data_content}") try: event_json = json.loads(data_content) logging.debug(f"Parsed JSON: {json.dumps(event_json, indent=2)}") # Filter by severity severity = event_json.get('severity') if severity not in trigger_severities: logging.info(f"Event severity '{severity}' does not match trigger severities: {trigger_severities}. Skipping.") continue # Extract client IPs from the parsed JSON client_ips = event_json.get('clientIPs', []) if client_ips: logs.extend(client_ips) logging.info(f"Extracted Client IPs: {client_ips}") except json.JSONDecodeError as e: logging.error(f"Failed to parse JSON from event.Data: {e}") except Exception as e: logging.warning(f"Failed to process log entry: {e}") # Update the latest processed record info latest_record = max(latest_record, record_number) latest_event_time = event_time latest_event_id = event_id finally: win32evtlog.CloseEventLog(h) # Save the latest processed record info if latest_event_time and latest_event_id: save_last_processed_record(latest_record, latest_event_time, latest_event_id) logging.info(f"Total Extracted Client IPs: {len(logs)} - {logs}") return logs # Flags to enable/disable features enable_isolation = True # Set this to False to disable isolation enable_broadcast = False # Set this to False to disable broadcast messages enable_full_scan = False # Set this to False to disable full disk scan # SentinelOne instance URL and API token instance_url = "https://xxxxx.sentinelone.net" api_token = "yyyyyyyy" # Headers including Authorization token headers = { "Authorization": f"Bearer {api_token}", "Content-Type": "application/json" } def get_device_id_by_ip(host_ip): # Step 1: Get a list of all agents and their IP addresses query_agents_url = f"{instance_url}/web/api/v2.1/agents" try: response = requests.get(query_agents_url, headers=headers) if response.status_code == 200: agents = response.json().get("data", []) # Search for the agent ID with the given IP address matching_agents = [agent["id"] for agent in agents if host_ip in [ip for ni in agent.get("networkInterfaces", []) for ip in ni.get("inet", [])]] if matching_agents: return matching_agents[0] # Return the first matching agent ID else: logging.error(f"No agents found with IP address {host_ip}") return None else: logging.error(f"Failed to query agents: {response.status_code} - {response.text}") return None except Exception as e: logging.error(f"An error occurred while querying agents: {str(e)}") return None def isolate_host(device_id): # Step 2: Isolate the host based on the agent ID if device_id: isolate_url = f"{instance_url}/web/api/v2.1/agents/actions/disconnect" payload = { "filter": { "ids": [device_id] } } try: isolate_response = requests.post(isolate_url, json=payload, headers=headers) if isolate_response.status_code == 200: logging.info(f"Host isolated successfully: {isolate_response.json()}") return isolate_response.json() else: logging.error(f"Failed to isolate host: {isolate_response.status_code} - {isolate_response.text}") return None except Exception as e: logging.error(f"An error occurred while isolating the host: {str(e)}") return None else: logging.error("No valid device ID provided for isolation.") return None def broadcast_message(device_id, message): # Step 3: Broadcast a message to the host based on the agent ID if device_id and len(message) <= 140: broadcast_url = f"{instance_url}/web/api/v2.1/agents/actions/broadcast" payload = { "filter": { "ids": [device_id] }, "data": { "message": message } } try: broadcast_response = requests.post(broadcast_url, json=payload, headers=headers) if broadcast_response.status_code == 200: logging.info(f"Message broadcasted successfully: {broadcast_response.json()}") return broadcast_response.json() else: logging.error(f"Failed to broadcast message: {broadcast_response.status_code} - {broadcast_response.text}") return None except Exception as e: logging.error(f"An error occurred while broadcasting the message: {str(e)}") return None else: logging.error("No valid device ID or message too long.") return None def initiate_full_scan(device_id): # Step 4: Initiate a full disk scan on the host based on the agent ID if device_id: scan_url = f"{instance_url}/web/api/v2.1/agents/actions/initiate-scan" payload = { "filter": { "ids": [device_id] } } try: scan_response = requests.post(scan_url, json=payload, headers=headers) if scan_response.status_code == 200: logging.info(f"Full scan initiated successfully: {scan_response.json()}") return scan_response.json() else: logging.error(f"Failed to initiate full scan: {scan_response.status_code} - {scan_response.text}") return None except Exception as e: logging.error(f"An error occurred while initiating the full scan: {str(e)}") return None else: logging.error("No valid device ID provided for full scan.") return None def main(): """Main execution logic.""" setup_logging() # Collect client IPs from Windows Event Logs client_ips = collect_windows_event_log(server, log_name, source_names) if not client_ips: logging.warning("No client IPs found in Windows Event Logs.") return # Process each client IP and perform SentinelOne actions for client_ip in client_ips: try: # Step 1: Get device ID by IP address device_id = get_device_id_by_ip(client_ip) if not device_id: logging.warning(f"Device ID not found for IP {client_ip}. Skipping.") continue # Step 2: Perform actions based on flags if enable_isolation: isolate_host(device_id) if enable_broadcast: message = "Security alert: Please check your system." broadcast_message(device_id, message) if enable_full_scan: initiate_full_scan(device_id) except Exception as e: logging.error(f"Failed to process actions for IP {client_ip}: {e}") if __name__ == "__main__": main()