Autonomous Multi-Agent SOC & Threat Hunting
Problem Statement
Ransomware propagation happens within minutes. Human-driven triage across network firewalls, Active Directory, and endpoint agents cannot act quickly enough to prevent domain-wide encryption. Multi-agent teams with specialized roles (Network Specialist, Endpoint Specialist, Identity Specialist) are needed to collaborate autonomously to isolate compromised hosts and revoke active sessions simultaneously.
Goal
Implement a CrewAI multi-agent team that coordinates concurrent containment actions across Network, Host, and Identity boundaries during a ransomware outbreak.
Outcome Measurement
- Response Time: Outbreak containment execution reduced to under 5 seconds.
- Blast Radius Reduction: 98% reduction in lateral spread across simulated subnets.
How the Code was Built
Uses crewai to instantiate three distinct agents (EndpointAgent, NetworkAgent, IdentityAgent) with assigned tasks collaborating through a centralized Crew manager.
Executable Code
from crewai import Agent, Task, Crew, Processfrom langchain_openai import ChatOpenAIllm = ChatOpenAI(model="gpt-4o", temperature=0)endpoint_agent = Agent( role="Endpoint Security Specialist", goal="Identify infected hosts and issue host isolation commands.", backstory="You are an expert EDR operator specializing in isolating compromised Windows servers.", verbose=True, llm=llm)network_agent = Agent( role="Network Security Operator", goal="Block command and control (C2) IPs on perimeter firewalls.", backstory="You manage edge firewalls and can instantly push rule blocks to stop data exfiltration.", verbose=True, llm=llm)identity_agent = Agent( role="Identity & Access Specialist", goal="Revoke compromised user refresh tokens and disable Active Directory accounts.", backstory="You enforce IAM policy and can terminate active OAuth sessions across corporate SSO.", verbose=True, llm=llm)task1 = Task( description="Analyze host alert for WORKSTATION-01 (Ransomware extension detected). Plan isolation payload.", expected_output="Host containment command dictionary.", agent=endpoint_agent)task2 = Task( description="Identify C2 IP associated with host alert (198.51.100.99) and plan edge firewall block rule.", expected_output="Firewall rule JSON snippet.", agent=network_agent)task3 = Task( description="Revoke session tokens for compromised user 'user_finance_01'.", expected_output="IAM token revocation confirmation.", agent=identity_agent)ransomware_crew = Crew( agents=[endpoint_agent, network_agent, identity_agent], tasks=[task1, task2, task3], process=Process.sequential)if __name__ == "__main__": result = ransomware_crew.kickoff() print("\n--- RANSOMWARE CONTAINMENT CREW EXECUTION RESULT ---") print(result)






Leave a Reply