LLMs for Security Operations

pip install anthropic
# Set your API key — never hardcode it
export ANTHROPIC_API_KEY=sk-ant-your-key-here

import anthropic
import pandas as pd
import json
import os

client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

SYSTEM_PROMPT = """You are a senior security analyst at a mid-sized technology company.
You receive anomaly detection alerts and produce structured investigation briefs.

Your company context:
- 200 employees, primarily engineering and operations
- Users log in from US cities; international logins are rare and suspicious
- Business hours are 7am-8pm local time; after-hours logins need explanation
- Normal login volume per user: 3-10 per day

For each alert, respond ONLY with a JSON object in this exact format:
{
  "severity": "CRITICAL|HIGH|MEDIUM|LOW",
  "likely_cause": "1-2 sentence assessment of what probably happened",
  "confidence": "HIGH|MEDIUM|LOW",
  "immediate_actions": ["action 1", "action 2", "action 3"],
  "escalate_to_manager": true|false,
  "escalate_reason": "why or why not",
  "investigation_steps": ["step 1", "step 2", "step 3"]
}

Do not include any text outside the JSON object."""

def triage_alert(alert: dict) -> dict:
    user_message = f"""Alert Details:
Type: {alert['type']}
User: {alert['user']}
Date: {alert['date']}
Detail: {alert['detail']}
Severity (detection system): {alert['severity']}

Analyze this alert and produce an investigation brief."""

    response = client.messages.create(
        model="claude-opus-4-5-20251101",
        max_tokens=1024,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": user_message}]
    )

    try:
        brief = json.loads(response.content[0].text)
        brief["alert"] = alert  # Attach original for reference
        return brief
    except json.JSONDecodeError:
        return {"error": "Parse failed", "raw": response.content[0].text, "alert": alert}

def run_triage():
    # Load alerts from Day 2
    alerts_df = pd.read_csv("alerts.csv")
    alerts = alerts_df.to_dict("records")

    print(f"Processing {len(alerts)} alerts...\n")
    briefs = []

    for i, alert in enumerate(alerts, 1):
        print(f"[{i}/{len(alerts)}] Triaging {alert['type']} for {alert['user']}...")
        brief = triage_alert(alert)
        briefs.append(brief)

        # Print summary to console
        if "error" not in brief:
            escalate = "ESCALATE" if brief.get("escalate_to_manager") else "monitor"
            print(f"  → {brief['severity']} | {escalate} | {brief['likely_cause'][:80]}...")
        print()

    # Save full briefs to JSON
    with open("investigation_briefs.json", "w") as f:
        json.dump(briefs, f, indent=2, default=str)

    # Print critical/high escalations
    print("\n=== ESCALATE NOW ===")
    for brief in briefs:
        if brief.get("escalate_to_manager") and "error" not in brief:
            alert = brief["alert"]
            print(f"\nUSER: {alert['user']} | {alert['type']}")
            print(f"CAUSE: {brief['likely_cause']}")
            print("IMMEDIATE ACTIONS:")
            for action in brief.get("immediate_actions", []):
                print(f"  • {action}")

    print(f"\nFull briefs saved to investigation_briefs.json")

if __name__ == "__main__":
    run_triage()

def analyze_user_timeline(user: str, alerts: list, log_df: pd.DataFrame) -> str:
    """Analyze all alerts for one user with full context."""
    user_alerts = [a for a in alerts if a["user"] == user]
    user_logs = log_df[log_df["user"] == user].tail(50)

    # Summarize recent log activity
    log_summary = user_logs[["timestamp","result","location","ip"]].to_string(index=False)

    prompt = f"""Analyze this user's complete activity timeline and provide a threat assessment.

User: {user}
Active Alerts: {json.dumps(user_alerts, default=str)}

Recent Activity Log (last 50 events):
{log_summary}

Provide:
1. Overall threat level (CRITICAL/HIGH/MEDIUM/LOW/BENIGN)
2. Most likely scenario (3-4 sentences)
3. Top 3 immediate actions
4. Whether to disable account now"""

    response = client.messages.create(
        model="claude-opus-4-5-20251101",
        max_tokens=1200,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text