Threat Intelligence & SIEM Pipeline Agents
Problem Statement
Honeytokens (synthetic credentials, AWS API keys, or fake document files placed as traps) generate alerts when accessed by attackers. However, security teams need immediate contextual triage when a canary trips to determine if an attacker is inside the network or if an internal developer accidentally stumbled upon the token.
Goal
Deploy an agent that processes honeytoken trigger alerts, traces the source IP/identity, and coordinates immediate deception response protocols.
Outcome Measurement
- Intruder Detection: 100% true-positive alert rate on honeytoken triggers.
- Triage Acceleration: Intruder IP isolated within 3 seconds of token access.
How the Code was Built
Ingests canary token telemetry payload and generates immediate SOC containment directives.
Executable Code
from pydantic import BaseModel, Fieldfrom langchain_openai import ChatOpenAIclass CanaryTriggerResponse(BaseModel): threat_severity: str = "CRITICAL" attacker_ip: str compromised_canary: str containment_instructions: list[str]def process_canary_alert(alert_payload: dict) -> CanaryTriggerResponse: llm = ChatOpenAI(model="gpt-4o", temperature=0) structured_llm = llm.with_structured_output(CanaryTriggerResponse) prompt = f"Analyze this honeytoken deployment trigger alert and issue containment instructions:\n{alert_payload}" return structured_llm.invoke(prompt)if __name__ == "__main__": canary_alert = { "token_type": "AWS API Access Key", "token_id": "AKIA_FAKE_CANARY_99", "source_ip": "198.51.100.77", "user_agent": "Boto3/1.26.0 Python/3.10", "requested_action": "sts:GetCallerIdentity", "location": "Planted in GitHub Repo /dev-internal-private" } res = process_canary_alert(canary_alert) print(f"Severity: {res.threat_severity}") print(f"Attacker IP: {res.attacker_ip}") print("Containment Steps:", res.containment_instructions)
Leave a Reply