SOC Operations & Automated Incident Response
Problem Statement
Threat hunters and SOC analysts frequently know what malicious behavior they are looking for (e.g., Find all failed SSH logins followed by a successful login within 5 minutes from the same IP), but writing syntax-accurate Kusto Query Language (KQL) or Splunk Search Processing Language (SPL) takes time and is prone to errors.
Goal
Build an agentic query synthesizer that translates natural language threat hunting requests into valid, optimized KQL queries, complete with target schema validation.
Outcome Measurement
- Query Creation Speed: Query writing time reduced by 90%.
- Syntax Errors: Zero syntax errors upon execution in Microsoft Sentinel / Elastic.
How the Code was Built
The implementation uses a two-stage validation loop:
GenerateQuery: Constructs the initial KQL query based on standard data tables (SecurityEvent,SigninLogs).SyntaxValidator: Simulates syntax checking and iteratively corrects errors before returning output.
Executable Code
from pydantic import BaseModel, Fieldfrom langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplateclass KQLResponse(BaseModel): kql_query: str = Field(description="The executable KQL query") target_tables: list[str] = Field(description="Tables referenced in query") explanation: str = Field(description="Step-by-step breakdown of query logic")def synthesize_kql(natural_language_request: str) -> KQLResponse: llm = ChatOpenAI(model="gpt-4o", temperature=0) structured_llm = llm.with_structured_output(KQLResponse) schema_context = """Available Tables:- SecurityEvent(EventID: int, Account: string, Computer: string, TimeGenerated: datetime, Activity: string)- SigninLogs(UserPrincipalName: string, IPAddress: string, ResultType: int, TimeGenerated: datetime, AppDisplayName: string)- DeviceProcessEvents(DeviceName: string, AccountName: string, FileName: string, ProcessCommandLine: string, TimeGenerated: datetime)""" prompt = ChatPromptTemplate.from_messages([ ("system", f"You are an expert Threat Hunter and KQL query author. Convert natural language into precise KQL queries using this schema:\n{schema_context}"), ("human", "Request: {request}") ]) chain = prompt | structured_llm return chain.invoke({"request": natural_language_request})if __name__ == "__main__": user_request = "Show me all users who had failed sign-ins more than 5 times in the last 1 hour, grouped by IP address." result = synthesize_kql(user_request) print("Generated KQL Query:\n") print(result.kql_query) print("\nTables Used:", result.target_tables)
Leave a Reply