agentic-soc-platform

# Agentic SOC Platform Skill > Skill by [ara.so](https://ara.so) — AI Agent Skills collection ## Overview Agentic SOC Platform (ASP) is an open-source, AI-driven security operations automation platform that combines SIEM integration, AI agents (Langgraph/Dify), and a built-in Security Incident Response Platform (SIRP). It processes security alerts through Redis streams, enriches them with AI analysis, and enables automated response workflows. **Key Components:** - **Modules**: Streaming processors that consume alerts from Redis streams and perform AI-driven analysis - **Playbooks**: Event-driven automation tasks triggered manually from the SIRP UI - **SIRP Platform**: Built on Nocoly for case management, alerts, and artifacts - **AI Agents**: Support for Langgraph, Dify, and local LLMs ## Installation ### Docker Deployment (Recommended) ```bash # Clone repository git clone https://github.com/FunnyWolf/agentic-soc-platform.git cd agentic-soc-platform # Start with Docker Compose cd Docker docker-compose up -d # Services will be available at: # - SIRP Platform: http://localhost:8000 # - Redis: localhost:6379 # - Webhook Receiver: http://localhost:5000 ``` ### Manual Installation ```bash # Python 3.9+ required git clone https://github.com/FunnyWolf/agentic-soc-platform.git cd agentic-soc-platform # Install dependencies pip install -r requirements.txt # Configure environment cp .env.example .env # Edit .env with your settings # Initialize database python manage.py migrate # Start services python manage.py runserver # SIRP platform python module_engine.py # Module processor python playbook_loader.py # Playbook executor python webhook_receiver.py # Alert ingestion ``` ## Configuration ### Environment Variables ```bash # .env file REDIS_HOST=localhost REDIS_PORT=6379 REDIS_DB=0 # Database DATABASE_URL=postgresql://user:pass@localhost:5432/asp_db # AI Agent Configuration OPENAI_API_KEY=${OPENAI_API_KEY} OPENAI_API_BASE=https://api.openai.com/v1 # Dify Configuration DIFY_API_URL=http://localhost:5001 DIFY_API_KEY=${DIFY_API_KEY} # Local LLM (Ollama) OLLAMA_BASE_URL=http://localhost:11434 # SIRP Configuration SIRP_API_URL=http://localhost:8000 SIRP_API_KEY=${SIRP_API_KEY} # Webhook Settings WEBHOOK_PORT=5000 WEBHOOK_SECRET=${WEBHOOK_SECRET} ``` ### Redis Stream Configuration ```python # config/redis_streams.py ALERT_STREAMS = { 'edr_alerts': 'stream:edr:alerts', 'ndr_alerts': 'stream:ndr:alerts', 'siem_alerts': 'stream:siem:alerts', 'email_threats': 'stream:email:threats', } CONSUMER_GROUPS = { 'edr_analyzer': ['stream:edr:alerts'], 'ndr_analyzer': ['stream:ndr:alerts'], 'threat_enricher': ['stream:siem:alerts', 'stream:email:threats'], } ``` ## Core Architecture ### Alert Processing Flow ```python # webhook_receiver.py - Receiving alerts from SIEM from flask import Flask, request, jsonify import redis import json app = Flask(__name__) r = redis.Redis(host='localhost', port=6379, db=0) @app.route('/webhook/alert', methods=['POST']) def receive_alert(): """Receive alert from SIEM and push to Redis stream""" alert_data = request.json # Determine stream based on alert source source = alert_data.get('source', 'unknown') stream_key = f"stream:{source}:alerts" # Push to Redis stream message_id = r.xadd( stream_key, { 'alert_id': alert_data.get('id'), 'payload': json.dumps(alert_data), 'timestamp': alert_data.get('timestamp'), 'severity': alert_data.get('severity', 'medium') } ) return jsonify({ 'status': 'success', 'stream': stream_key, 'message_id': message_id.decode() }), 200 if __name__ == '__main__': app.run(host='0.0.0.0', port=5000) ``` ## Creating Modules Modules are streaming processors that consume alerts from Redis streams. ### Basic Module Structure ```python # modules/edr_analyzer.py from core.module_base import ModuleBase from agents.langgraph_agent import LanggraphAgent import json class EDRAnalyzer(ModuleBase): """Analyze EDR alerts using AI agent""" def __init__(self): super().__init__( name='edr_analyzer', streams=['stream:edr:alerts'], consumer_group='edr_analysis_group' ) self.agent = LanggraphAgent( model='gpt-4', system_prompt="""You are a security analyst specializing in EDR alerts. Analyze the alert and determine: 1. Is this a true positive or false positive? 2. What is the MITRE ATT&CK technique? 3. What is the recommended response? """ ) async def process_message(self, message_id, data): """Process individual alert from stream""" alert_payload = json.loads(data['payload']) # AI analysis analysis = await self.agent.analyze({ 'alert_type': alert_payload.get('type'), 'process_name': alert_payload.get('process_name'), 'command_line': alert_payload.get('command_line'), 'user': alert_payload.get('user'), 'host': alert_payload.get('host') }) # Create SIRP case sirp_case = await self.create_sirp_case({ 'title': f"EDR Alert: {alert_payload.get('type')}", 'severity': self._map_severity(analysis['confidence']), 'description': analysis['summary'], 'mitre_technique': analysis.get('mitre_technique'), 'recommended_action': analysis.get('recommendation'), 'artifacts': [ { 'type': 'process', 'value': alert_payload.get('process_name') }, { 'type': 'host', 'value': alert_payload.get('host') } ] }) # Acknowledge message await self.ack_message(message_id) return sirp_case def _map_severity(self, confidence): """Map AI confidence to severity level""" if confidence > 0.8: return 'critical' elif confidence > 0.6: return 'high' elif confidence > 0.4: return 'medium' return 'low' ``` ### Module Base Class ```python # core/module_base.py import redis import asyncio from abc import ABC, abstractmethod from typing import List, Dict import logging class ModuleBase(ABC): """Base class for all ASP modules""" def __init__(self, name: str, streams: List[str], consumer_group: str): self.name = name self.streams = streams self.consumer_group = consumer_group self.redis_client = redis.Redis( host='localhost', port=6379, db=0, decode_responses=True ) self.logger = logging.getLogger(f"module.{name}") self._setup_consumer_groups() def _setup_consumer_groups(self): """Create consumer groups for streams""" for stream in self.streams: try: self.redis_client.xgroup_create( stream, self.consumer_group, id='0', mkstream=True ) except redis.exceptions.ResponseError as e: if 'BUSYGROUP' not in str(e): raise async def run(self): """Main processing loop""" self.logger.info(f"Starting module {self.name}") while True: for stream in self.streams: messages = self.redis_client.xreadgroup( self.consumer_group, self.name, # Consumer name {stream: '>'}, count=10, block=1000 ) for stream_name, stream_messages in messages: for message_id, data in stream_messages: try: await self.process_message(message_id, data) except Exception as e: self.logger.error(f"Error processing {message_id}: {e}") await asyncio.sleep(0.1) @abstractmethod async def process_message(self, message_id: str, data: Dict): """Process individual message - must be implemented by subclass""" pass async def ack_message(self, message_id: str): """Acknowledge processed message""" for stream in self.streams: self.redis_client.xack(stream, self.consumer_group, message_id) async def create_sirp_case(self, case_data: Dict): """Create case in SIRP platform""" from clients.sirp_client import SIRPClient client = SIRPClient() return await client.create_case(case_data) ``` ## Creating Playbooks Playbooks are event-driven automation tasks triggered from the SIRP UI. ### Basic Playbook Structure ```python # playbooks/threat_intel_enrichment.py from core.playbook_base import PlaybookBase from typing import Dict, List class ThreatIntelEnrichment(PlaybookBase): """Enrich indicators with threat intelligence""" metadata = { 'name': 'Threat Intel Enrichment', 'description': 'Query VirusTotal, AbuseIPDB, and other TI sources', 'input_types': ['ip', 'domain', 'hash', 'url'], 'output_type': 'enrichment_report' } async def execute(self, artifact: Dict) -> Dict: """Execute playbook on artifact""" artifact_type = artifact['type'] artifact_value = artifact['value'] results = {} if artifact_type == 'ip': results['virustotal'] = await self._query_virustotal_ip(artifact_value) results['abuseipdb'] = await self._query_abuseipdb(artifact_value) results['shodan'] = await self._query_shodan(artifact_value) elif artifact_type == 'domain': results['virustotal'] = await self._query_virustotal_domain(artifact_value) results['urlscan'] = await self._query_urlscan(artifact_value) elif artifact_type == 'hash': results['virustotal'] = await self._query_virustotal_hash(artifact_value) results['hybrid_analysis'] = await self._query_hybrid_analysis(artifact_value) # Update artifact in SIRP await self.update_artifact(artifact['id'], { 'enrichment': results, 'reputation_score': self._calculate_reputation(results), 'tags': self._extract_tags(results) }) return { 'status': 'success', 'artifact_id': artifact['id'], 'enrichment_data': results } async def _query_virustotal_ip(self, ip: str) -> Dict: """Query VirusTotal IP endpoint""" import aiohttp import os api_key = os.getenv('VIRUSTOTAL_API_KEY') url = f"https://www.virustotal.com/api/v3/ip_addresses/{ip}" async with aiohttp.ClientSession() as session: async with session.get( url, headers={'x-apikey': api_key} ) as response: if response.status == 200: data = await response.json() return { 'malicious': data['data']['attributes']['last_analysis_stats']['malicious'], 'reputation': data['data']['attributes'].get('reputation', 0), 'country': data['data']['attributes'].get('country'), 'asn': data['data']['attributes'].get('asn') } return {'error': f"Status {response.status}"} async def _query_abuseipdb(self, ip: str) -> Dict: """Query AbuseIPDB""" import aiohttp import os api_key = os.getenv('ABUSEIPDB_API_KEY') url = 'https://api.abuseipdb.com/api/v2/check' async with aiohttp.ClientSession() as session: async with session.get( url, headers={'Key': api_key}, params={'ipAddress': ip, 'maxAgeInDays': 90} ) as response: if response.status == 200: data = await response.json() return {