From ea7984ce12f422ade8180c0055942643c6fff92e Mon Sep 17 00:00:00 2001 From: rootkit Date: Thu, 5 Feb 2026 02:35:12 +0800 Subject: [PATCH] tmp --- .gemini/GEMINI.md | 4 +- .github/copilot-instructions.md | 8 ++++ CLAUDE.md | 4 +- PLUGINS/SIEM/main.py | 40 ++++++++-------- PLUGINS/SIEM/models.py | 83 ++++++++++++++++++++++++++------- PLUGINS/SIEM/tools.py | 74 +++++++++++++++++++++++++++-- 6 files changed, 171 insertions(+), 42 deletions(-) create mode 100644 .github/copilot-instructions.md diff --git a/.gemini/GEMINI.md b/.gemini/GEMINI.md index bfde4d8..547fe25 100644 --- a/.gemini/GEMINI.md +++ b/.gemini/GEMINI.md @@ -3,4 +3,6 @@ - 每次生成代码前,需要描述方案和代码结构,等待用户确认后再生成代码 - 除非用户明确要求,否则代码中不要添加任何注释 - 除非用户明确要求,否则不要生成和添加任何文档 -- 不要运行任何测试,不要添加任何测试脚本和代码,不要进行语法检查 \ No newline at end of file +- 不要运行任何测试,不要添加任何测试脚本和代码,不要进行语法检查 +- 不要进行任何语法检查 +- 不要生成任何总结性内容和报告 \ No newline at end of file diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md new file mode 100644 index 0000000..547fe25 --- /dev/null +++ b/.github/copilot-instructions.md @@ -0,0 +1,8 @@ +# 必须遵守的规则 + +- 每次生成代码前,需要描述方案和代码结构,等待用户确认后再生成代码 +- 除非用户明确要求,否则代码中不要添加任何注释 +- 除非用户明确要求,否则不要生成和添加任何文档 +- 不要运行任何测试,不要添加任何测试脚本和代码,不要进行语法检查 +- 不要进行任何语法检查 +- 不要生成任何总结性内容和报告 \ No newline at end of file diff --git a/CLAUDE.md b/CLAUDE.md index 5d2be4e..6acc714 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -4,5 +4,5 @@ - 除非用户明确要求,否则代码中不要添加任何注释 - 除非用户明确要求,否则不要生成和添加任何文档 - 不要运行任何测试,不要添加任何测试脚本和代码,不要进行语法检查 - -# 代码说明 +- 不要进行任何语法检查 +- 不要生成任何总结性内容和报告 diff --git a/PLUGINS/SIEM/main.py b/PLUGINS/SIEM/main.py index 0f7a380..129a066 100644 --- a/PLUGINS/SIEM/main.py +++ b/PLUGINS/SIEM/main.py @@ -28,7 +28,7 @@ def main(): time_range_end=time_range_end, filters={ "event.outcome": "success", - "user.name": "user_002" + # "user.name": "user_002" } ) @@ -41,25 +41,25 @@ def main(): # 验证返回数据 print(f"Sample: {result.records[0]}") - query_input = AdaptiveQueryInput( - index_name="siem-network-traffic", - time_range_start=time_range_start, - time_range_end=time_range_end, - filters={ - "event.dataset": "network", - "destination.ip": "104.21.11.22", - "event.action": "deny" - } - ) - - result = toolkit.execute_adaptive_query(query_input) - print(f"Using time field: {query_input.time_field}") - print(f"Status: {result.status}") - print(f"Total Hits: {result.total_hits}") - - if result.records: - # 验证返回数据 - print(f"Sample: {result.records[0]}") + # query_input = AdaptiveQueryInput( + # index_name="siem-network-traffic", + # time_range_start=time_range_start, + # time_range_end=time_range_end, + # filters={ + # "event.dataset": "network", + # "destination.ip": "104.21.11.22", + # "event.action": "deny" + # } + # ) + # + # result = toolkit.execute_adaptive_query(query_input) + # print(f"Using time field: {query_input.time_field}") + # print(f"Status: {result.status}") + # print(f"Total Hits: {result.total_hits}") + # + # if result.records: + # # 验证返回数据 + # print(f"Sample: {result.records[0]}") if __name__ == "__main__": diff --git a/PLUGINS/SIEM/models.py b/PLUGINS/SIEM/models.py index bec2d92..5892fa1 100644 --- a/PLUGINS/SIEM/models.py +++ b/PLUGINS/SIEM/models.py @@ -8,35 +8,52 @@ from pydantic import BaseModel, Field, field_validator class SchemaExplorerInput(BaseModel): target_index: Optional[str] = Field( default=None, - description="If None, lists all indices. If provided, details fields." + description=( + "Target index to explore. " + "If None: returns a list of all available indices with descriptions (list of dicts with 'name' and 'description'). " + "If provided: returns detailed field metadata for that specific index (list of field schemas with 'name', 'type', 'description', etc.)" + ) ) class AdaptiveQueryInput(BaseModel): - index_name: str = Field(..., description="Target index") + index_name: str = Field( + ..., + description="Target SIEM index/source name. Examples: 'logs-security', 'main', 'logs-endpoint'" + ) - # 新增：允许指定用于过滤的时间字段，默认为 @timestamp time_field: str = Field( default="@timestamp", - description="The field to apply time range filter on (e.g., 'event.created', '@timestamp'). Must be a Date type in SIEM." + description=( + "The field to apply time range filter on. " + "Commonly used fields: '@timestamp', 'event.created', '_time'. " + "Must be a Date/DateTime type in your SIEM." + ) ) time_range_start: str = Field( ..., - description="Start time in UTC ISO8601 format, e.g., '2026-02-04T06:00:00Z'" + description="Start time in UTC ISO8601 format. Format: 'YYYY-MM-DDTHH:MM:SSZ'. Example: '2026-02-04T06:00:00Z'" ) time_range_end: str = Field( ..., - description="End time in UTC ISO8601 format, e.g., '2026-02-04T07:00:00Z'" + description="End time in UTC ISO8601 format. Format: 'YYYY-MM-DDTHH:MM:SSZ'. Example: '2026-02-04T07:00:00Z'" ) filters: Dict[str, str] = Field( default_factory=dict, - description="Key-value pairs for exact matching (term query) e.g., {'event.outcome': 'success', 'source.ip': '45.33.22.11'}" + description=( + "Key-value pairs for exact matching filters (term/exact match, not full-text search). " + "Example: {'event.outcome': 'success', 'source.ip': '45.33.22.11'}" + ) ) aggregation_fields: List[str] = Field( default_factory=list, - description="Fields to get statistics for. If empty, uses default key fields. e.g., ['event.outcome','source.ip'] " + description=( + "Fields to get top-N statistics for. " + "If empty, uses backend-specific default key fields. " + "Example: ['event.outcome', 'source.ip', 'process.name']" + ) ) @field_validator('time_range_start', 'time_range_end') @@ -51,15 +68,49 @@ class AdaptiveQueryInput(BaseModel): return v -# --- Output Models (保持不变) --- +# --- Output Models --- class FieldStat(BaseModel): - field_name: str - top_values: Dict[str, int] + field_name: str = Field( + ..., + description="Name of the field for which statistics are computed" + ) + top_values: Dict[str, int] = Field( + ..., + description="Top-N value distribution for the field (key: value, int: count)" + ) class AdaptiveQueryOutput(BaseModel): - status: str = Field(..., description="summary | sample | full") - total_hits: int - message: str - statistics: List[FieldStat] - records: List[Dict[str, Any]] + status: str = Field( + ..., + description=( + "Response type indicator based on result volume. " + "Possible values: 'full' (complete logs, < 20 results), " + "'sample' (statistics + sample records, 20-1000 results), " + "'summary' (statistics only, > 1000 results)" + ) + ) + total_hits: int = Field( + ..., + description="Total number of matching records in the SIEM backend" + ) + message: str = Field( + ..., + description="Human-readable status message describing the response" + ) + statistics: List[FieldStat] = Field( + ..., + description=( + "Top-N value distribution for each aggregation field. " + "Each FieldStat contains field_name and top_values (dict mapping values to their counts)" + ) + ) + records: List[Dict[str, Any]] = Field( + ..., + description=( + "Actual log records returned based on status: " + "'full' status returns all records up to SAMPLE_THRESHOLD; " + "'sample' status returns first 3 representative records; " + "'summary' status returns empty list" + ) + ) diff --git a/PLUGINS/SIEM/tools.py b/PLUGINS/SIEM/tools.py index 644035a..5971335 100644 --- a/PLUGINS/SIEM/tools.py +++ b/PLUGINS/SIEM/tools.py @@ -18,8 +18,29 @@ SAMPLE_THRESHOLD = 20 class SIEMToolKit(object): + @classmethod def explore_schema(cls, input_data: SchemaExplorerInput = SchemaExplorerInput(target_index=None)): + """ + Explore available SIEM indices and their field schemas. + + This tool helps agents discover what data sources are available and what fields they contain. + It supports two modes based on the target_index parameter in SchemaExplorerInput: + 1. List all indices (when target_index is None) + 2. Get detailed field information for a specific index + + See SchemaExplorerInput for detailed parameter documentation. + + Raises: + ValueError: If the specified target_index is not found in the registry. + + Example Usage by Agent: + # List all indices + explore_schema() + + # Get details on "logs-security" index + explore_schema(SchemaExplorerInput(target_index="logs-security")) + """ try: if not input_data.target_index: # Agent 看到的是统一的列表，不关心 Backend @@ -39,6 +60,35 @@ class SIEMToolKit(object): @classmethod def execute_adaptive_query(cls, input_data: AdaptiveQueryInput) -> AdaptiveQueryOutput: + """ + Execute adaptive queries against SIEM backends (ELK or Splunk) with intelligent response formatting. + + This tool executes queries with automatic backend detection and response optimization: + - Automatically adjusts response format based on result volume: + * Full logs: Complete log records (for < 20 results) + * Sample: Statistics + sample records (for 20-1000 results) + * Summary: Statistics only (for > 1000 results) + - Provides top-N statistics for specified aggregation fields + - Handles time range filtering with UTC ISO8601 timestamps + + Raises: + ValueError: If time format is invalid or backend is unsupported + ConnectionError: If SIEM backend is unreachable + + Example Usage by Agent: + # Query security logs from last hour + input_data = AdaptiveQueryInput( + index_name="logs-security", + time_range_start="2026-02-04T06:00:00Z", + time_range_end="2026-02-04T07:00:00Z", + filters={"event.outcome": "failure"}, + aggregation_fields=["event.action", "user.name"] + ) + result = execute_adaptive_query(input_data) + + # Agent can then analyze result.statistics for patterns + # and if needed, drill down with result.records + """ # --- 路由层 (Router Layer) --- backend = get_backend_type(input_data.index_name) @@ -67,7 +117,8 @@ class SIEMToolKit(object): query_body = {"bool": {"must": must_clauses}} agg_fields = input_data.aggregation_fields or get_default_agg_fields(input_data.index_name) - aggs_dsl = {f: {"terms": {"field": f, "size": 5}} for f in agg_fields} + # Build aggregations using .keyword subfields to avoid fielddata disabled issues + aggs_dsl = cls._build_safe_aggs(agg_fields) response = client.search( index=input_data.index_name, query=query_body, aggs=aggs_dsl, size=3, track_total_hits=True @@ -79,8 +130,10 @@ class SIEMToolKit(object): stats_output = [] if "aggregations" in response: for field in agg_fields: - if field in response["aggregations"]: - buckets = response["aggregations"][field]["buckets"] + # The aggregation key uses .keyword suffix, but we report original field name + agg_key = f"{field}.keyword" + if agg_key in response["aggregations"]: + buckets = response["aggregations"][agg_key]["buckets"] stats_output.append(FieldStat( field_name=field, top_values={b["key"]: b["doc_count"] for b in buckets} @@ -192,6 +245,21 @@ class SIEMToolKit(object): records=final_records ) + # ========================================== + # Helper: Build Safe Aggregations (Handle fielddata disabled) + # ========================================== + @classmethod + def _build_safe_aggs(cls, agg_fields): + """ + Build aggregation DSL with keyword subfields to avoid fielddata issues. + For text fields without fielddata, use .keyword suffix. + """ + safe_aggs = {} + for f in agg_fields: + # Use keyword subfield for all fields to avoid fielddata disabled issues + safe_aggs[f"{f}.keyword"] = {"terms": {"field": f"{f}.keyword", "size": 5}} + return safe_aggs + # ========================================== # Helper: Shared Logic for ELK (Refactored) # ==========================================