AI Orchestration for Security Operations

from anthropic import Anthropic
from datetime import datetime
import json

class SecurityTriageAgent:
    """Single agent for automated alert triage and classification."""

    def __init__(self):
        self.client = Anthropic()
        self.severity_thresholds = {
            "critical": 9.0,
            "high": 7.0,
            "medium": 4.0,
            "low": 0.0
        }

    def triage_alert(self, alert: dict) -> dict:
        """Analyze and classify a security alert."""

        prompt = f"""Analyze this security alert and provide triage assessment.

ALERT DATA:
{json.dumps(alert, indent=2)}

Provide your analysis in JSON format:
{{
    "classification": "true_positive|false_positive|suspicious",
    "severity": "critical|high|medium|low",
    "confidence": 0.0-1.0,
    "attack_technique": "MITRE ATT&CK technique ID if applicable",
    "recommended_actions": ["list of immediate actions"],
    "enrichment_needed": ["list of additional data sources to query"],
    "reasoning": "brief explanation of classification"
}}"""

        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1000,
            messages=[{"role": "user", "content": prompt}]
        )

        return json.loads(response.content[0].text)

    def batch_triage(self, alerts: list[dict]) -> list[dict]:
        """Process multiple alerts with priority ordering."""
        results = []
        for alert in alerts:
            triage = self.triage_alert(alert)
            triage["original_alert"] = alert
            triage["triaged_at"] = datetime.utcnow().isoformat()
            results.append(triage)

        # Sort by severity and confidence
        severity_order = {"critical": 0, "high": 1, "medium": 2, "low": 3}
        results.sort(key=lambda x: (severity_order[x["severity"]], -x["confidence"]))

        return results

from enum import Enum
from dataclasses import dataclass
from typing import Callable
import asyncio

class AgentRole(Enum):
    TRIAGE = "triage"
    ENRICHMENT = "enrichment"
    THREAT_INTEL = "threat_intel"
    RESPONSE = "response"
    SUPERVISOR = "supervisor"

@dataclass
class AgentMessage:
    sender: AgentRole
    recipient: AgentRole
    content: dict
    priority: int = 0

class SecurityAgentOrchestrator:
    """Coordinate multiple specialized security agents."""

    def __init__(self):
        self.agents = {}
        self.message_queue = asyncio.Queue()
        self.investigation_state = {}

    def register_agent(self, role: AgentRole, handler: Callable):
        """Register a specialized agent handler."""
        self.agents[role] = handler

    async def investigate_incident(self, incident: dict) -> dict:
        """Orchestrate multi-agent incident investigation."""

        # Phase 1: Initial triage
        triage_result = await self.agents[AgentRole.TRIAGE](incident)

        # Phase 2: Parallel enrichment
        enrichment_tasks = [
            self.agents[AgentRole.ENRICHMENT](incident, triage_result),
            self.agents[AgentRole.THREAT_INTEL](incident, triage_result)
        ]
        enrichment_results = await asyncio.gather(*enrichment_tasks)

        # Phase 3: Supervisor synthesis
        combined_context = {
            "incident": incident,
            "triage": triage_result,
            "enrichment": enrichment_results[0],
            "threat_intel": enrichment_results[1]
        }

        # Phase 4: Response recommendation
        response = await self.agents[AgentRole.SUPERVISOR](combined_context)

        return response

from enum import Enum
from dataclasses import dataclass
import asyncio

class ApprovalLevel(Enum):
    NONE = "none"
    OPTIONAL = "optional"
    RECOMMENDED = "recommended"
    REQUIRED = "required"
    DUAL_REQUIRED = "dual_required"

@dataclass
class PendingAction:
    action_id: str
    action_type: str
    target: str
    reasoning: str
    approval_level: ApprovalLevel
    timeout_seconds: int
    approvals: list = None

class HumanInTheLoopController:
    """Manage human approval workflows for AI-recommended actions."""

    def __init__(self, notification_service, approval_store):
        self.notifications = notification_service
        self.approvals = approval_store
        self.action_policies = self._load_policies()

    async def request_approval(self, action: PendingAction) -> bool:
        """Request human approval for a security action."""

        if action.approval_level == ApprovalLevel.NONE:
            return True

        # Store pending action
        await self.approvals.store(action)

        # Notify appropriate approvers
        await self.notifications.send_approval_request(
            action=action,
            channels=["slack", "pagerduty", "email"]
        )

        # Wait for approval with timeout
        try:
            approved = await asyncio.wait_for(
                self._wait_for_approval(action),
                timeout=action.timeout_seconds
            )
            return approved
        except asyncio.TimeoutError:
            return self._handle_timeout(action)

    def _handle_timeout(self, action: PendingAction) -> bool:
        """Handle approval timeout based on policy."""
        if action.approval_level in [ApprovalLevel.REQUIRED, ApprovalLevel.DUAL_REQUIRED]:
            self.notifications.escalate(action)
            return False
        return action.approval_level == ApprovalLevel.OPTIONAL

from anthropic import Anthropic
import json

class AlertTriageOrchestrator:
    """Orchestrate alert triage across multiple data sources."""

    def __init__(self):
        self.client = Anthropic()
        self.enrichment_sources = []

    async def triage_with_enrichment(self, alert: dict) -> dict:
        """Full triage workflow with parallel enrichment."""

        # Parallel enrichment queries
        enrichment_data = await self._gather_enrichment(alert)

        # AI-powered classification with full context
        prompt = f"""You are a security analyst performing alert triage.

ALERT:
{json.dumps(alert, indent=2)}

ENRICHMENT DATA:
{json.dumps(enrichment_data, indent=2)}

Classify this alert considering:
1. Is this a true positive, false positive, or requires investigation?
2. What is the business impact if this is a real attack?
3. What MITRE ATT&CK techniques are indicated?
4. What immediate actions should be taken?

Respond in JSON format with classification, severity (critical/high/medium/low),
confidence (0-1), mitre_techniques, and recommended_actions."""

        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1000,
            messages=[{"role": "user", "content": prompt}]
        )

        result = json.loads(response.content[0].text)
        result["enrichment"] = enrichment_data
        return result

class IncidentInvestigationAgent:
    """Automated incident investigation with evidence collection."""

    def __init__(self, siem_client, edr_client, ti_client):
        self.siem = siem_client
        self.edr = edr_client
        self.threat_intel = ti_client
        self.client = Anthropic()

    async def investigate(self, incident_id: str, initial_indicators: list) -> dict:
        """Conduct comprehensive incident investigation."""

        investigation = {
            "incident_id": incident_id,
            "timeline": [],
            "affected_assets": set(),
            "iocs": set(),
            "ttps": [],
            "recommendations": []
        }

        # Phase 1: Expand IOCs through pivoting
        expanded_iocs = await self._pivot_on_indicators(initial_indicators)
        investigation["iocs"].update(expanded_iocs)

        # Phase 2: Query all relevant data sources
        siem_results = await self.siem.search(expanded_iocs, hours=72)
        edr_results = await self.edr.hunt(expanded_iocs)
        ti_matches = await self.threat_intel.lookup(expanded_iocs)

        # Phase 3: Build investigation timeline
        timeline = self._build_timeline(siem_results, edr_results)
        investigation["timeline"] = timeline

        # Phase 4: AI analysis and recommendation
        analysis = await self._analyze_investigation(investigation, ti_matches)

        return analysis

class ThreatHuntingAgent:
    """AI-assisted threat hunting across enterprise data."""

    def __init__(self):
        self.client = Anthropic()
        self.hunt_hypotheses = self._load_hunt_library()

    def generate_hunt_queries(self, hypothesis: str, data_sources: list) -> list:
        """Generate hunting queries for a given hypothesis."""

        prompt = f"""Generate threat hunting queries for this hypothesis:

HYPOTHESIS: {hypothesis}

AVAILABLE DATA SOURCES: {json.dumps(data_sources)}

For each data source, provide:
1. Query syntax appropriate for that platform
2. Expected indicators if the hypothesis is true
3. False positive considerations
4. MITRE ATT&CK technique coverage

Format as JSON array of hunt queries."""

        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2000,
            messages=[{"role": "user", "content": prompt}]
        )

        return json.loads(response.content[0].text)

    async def execute_hunt(self, queries: list) -> dict:
        """Execute hunting queries and analyze results."""
        results = []
        for query in queries:
            query_results = await self._execute_query(query)
            if query_results:
                analysis = await self._analyze_hunt_results(query, query_results)
                results.append(analysis)

        return {"findings": results, "summary": self._summarize_hunt(results)}

from dataclasses import dataclass
from enum import Enum
from typing import Optional

class ResponseAction(Enum):
    BLOCK_IP = "block_ip"
    ISOLATE_HOST = "isolate_host"
    DISABLE_USER = "disable_user"
    QUARANTINE_FILE = "quarantine_file"
    RESET_PASSWORD = "reset_password"
    REVOKE_SESSIONS = "revoke_sessions"

@dataclass
class ResponsePlan:
    actions: list[ResponseAction]
    targets: list[str]
    reasoning: str
    approval_required: bool
    rollback_plan: Optional[str] = None

class ResponseOrchestrator:
    """Execute automated response actions with safety controls."""

    def __init__(self, action_handlers: dict, approval_controller):
        self.handlers = action_handlers
        self.approvals = approval_controller
        self.executed_actions = []

    async def execute_response_plan(self, plan: ResponsePlan) -> dict:
        """Execute a response plan with appropriate approvals."""

        results = {"success": [], "failed": [], "pending_approval": []}

        for action, target in zip(plan.actions, plan.targets):
            # Check if approval is needed
            if plan.approval_required:
                approved = await self.approvals.request_approval(action, target)
                if not approved:
                    results["pending_approval"].append({"action": action, "target": target})
                    continue

            # Execute action with rollback capability
            try:
                result = await self.handlers[action](target)
                self.executed_actions.append({
                    "action": action,
                    "target": target,
                    "result": result,
                    "rollback": plan.rollback_plan
                })
                results["success"].append({"action": action, "target": target})
            except Exception as e:
                results["failed"].append({"action": action, "target": target, "error": str(e)})

        return results

from dataclasses import dataclass
from typing import Protocol
import httpx

class SecurityToolClient(Protocol):
    """Protocol for security tool integrations."""

    async def query(self, query: str) -> dict: ...
    async def execute_action(self, action: str, target: str) -> dict: ...

@dataclass
class ToolCredentials:
    api_key: str
    base_url: str
    timeout: int = 30

class SecurityToolOrchestrator:
    """Unified interface for security tool integrations."""

    def __init__(self):
        self.tools: dict[str, SecurityToolClient] = {}

    def register_tool(self, name: str, client: SecurityToolClient):
        self.tools[name] = client

    async def parallel_query(self, query: str, tools: list[str]) -> dict:
        """Execute query across multiple tools in parallel."""
        async with httpx.AsyncClient() as http_client:
            results = {}
            for tool_name in tools:
                if tool_name in self.tools:
                    results[tool_name] = await self.tools[tool_name].query(query)
            return results

from dataclasses import dataclass
from datetime import datetime, timedelta

@dataclass
class GuardrailConfig:
    max_actions_per_hour: int = 100
    max_concurrent_isolations: int = 5
    require_rollback_plan: bool = True
    high_impact_threshold: float = 0.8

class SafetyGuardrails:
    """Enforce safety constraints on AI agent actions."""

    def __init__(self, config: GuardrailConfig):
        self.config = config
        self.action_log = []
        self.permitted_actions = set()

    def validate_action(self, action: str, target: str, context: dict) -> tuple[bool, str]:
        """Validate action against all guardrails."""

        # Check action is permitted
        if action not in self.permitted_actions:
            return False, f"Action {action} not in permitted list"

        # Check rate limits
        recent_actions = self._count_recent_actions(hours=1)
        if recent_actions >= self.config.max_actions_per_hour:
            return False, "Rate limit exceeded"

        # Check blast radius
        if action == "isolate_host":
            current_isolations = self._count_active_isolations()
            if current_isolations >= self.config.max_concurrent_isolations:
                return False, "Maximum concurrent isolations reached"

        # Check rollback plan exists
        if self.config.require_rollback_plan and "rollback_plan" not in context:
            return False, "Rollback plan required but not provided"

        return True, "Validation passed"

import json
from datetime import datetime
import structlog

class AISecurityAuditLogger:
    """Comprehensive audit logging for AI security operations."""

    def __init__(self, log_backend):
        self.logger = structlog.get_logger()
        self.backend = log_backend

    def log_decision(self, decision: dict):
        """Log an AI decision with full context for audit."""
        audit_record = {
            "timestamp": datetime.utcnow().isoformat(),
            "decision_id": decision.get("id"),
            "decision_type": decision.get("type"),
            "input_context": decision.get("context"),
            "reasoning": decision.get("reasoning"),
            "output": decision.get("output"),
            "confidence": decision.get("confidence"),
            "model_version": decision.get("model"),
            "human_override": decision.get("overridden", False)
        }

        self.logger.info("ai_decision", **audit_record)
        self.backend.store(audit_record)

    def log_action(self, action: dict):
        """Log an executed action with rollback information."""
        action_record = {
            "timestamp": datetime.utcnow().isoformat(),
            "action_id": action.get("id"),
            "action_type": action.get("type"),
            "target": action.get("target"),
            "triggered_by": action.get("decision_id"),
            "result": action.get("result"),
            "rollback_available": action.get("rollback") is not None
        }

        self.logger.info("ai_action", **action_record)
        self.backend.store(action_record)