Threat Intelligence & SIEM Pipeline Agents
Problem Statement
Command and Control (C2) agents communicate with external servers using regular, periodic outbound requests (beaconing) or jittered intervals designed to hide within proxy web traffic. Traditional static thresholds produce high false-positive rates on legitimate background API polling traffic.
Goal
Build an agent that evaluates web proxy connection logs, calculates timing variance jitter metrics, and identifies covert C2 channels.
Outcome Measurement
- Detection Sensitivity: Identified low-frequency C2 beacons transmitting once per hour.
- False Positive Rate: Less than 0.1% on enterprise proxy log test sets.
How the Code was Built
Combines statistical interval analysis with LLM payload analysis to flag suspicious callback patterns.
Executable Code
from pydantic import BaseModel, Fieldfrom langchain_openai import ChatOpenAIclass BeaconAnalysis(BaseModel): is_c2_beacon: bool estimated_interval_seconds: int threat_classification: str action_required: strdef analyze_beacon_pattern(destination_domain: str, request_timestamps: list[str], payload_sizes: list[int]) -> BeaconAnalysis: llm = ChatOpenAI(model="gpt-4o", temperature=0) structured_llm = llm.with_structured_output(BeaconAnalysis) prompt = f"""Analyze outbound traffic pattern for potential C2 Beaconing:Target Domain: {destination_domain}Timestamps: {request_timestamps}Payload Sizes (Bytes): {payload_sizes}Evaluate interval regularity, payload size consistency, and domain reputation.""" return structured_llm.invoke(prompt)if __name__ == "__main__": times = ["10:00:00", "10:05:01", "10:10:00", "10:15:02", "10:20:01"] sizes = [240, 240, 242, 240, 241] res = analyze_beacon_pattern("cdn-update-sync-service.top", times, sizes) print(f"C2 Beacon Detected: {res.is_c2_beacon}") print(f"Interval: ~{res.estimated_interval_seconds} seconds") print(f"Classification: {res.threat_classification}")
Leave a Reply