tmp

.gemini/GEMINI.md

@@ -3,4 +3,6 @@
 - 每次生成代码前,需要描述方案和代码结构,等待用户确认后再生成代码
 - 除非用户明确要求,否则代码中不要添加任何注释
 - 除非用户明确要求,否则不要生成和添加任何文档
-- 不要运行任何测试,不要添加任何测试脚本和代码,不要进行语法检查
+- 不要运行任何测试,不要添加任何测试脚本和代码,不要进行语法检查
+- 不要进行任何语法检查
+- 不要生成任何总结性内容和报告

.github/copilot-instructions.md

@@ -0,0 +1,8 @@
+# 必须遵守的规则
+
+- 每次生成代码前,需要描述方案和代码结构,等待用户确认后再生成代码
+- 除非用户明确要求,否则代码中不要添加任何注释
+- 除非用户明确要求,否则不要生成和添加任何文档
+- 不要运行任何测试,不要添加任何测试脚本和代码,不要进行语法检查
+- 不要进行任何语法检查
+- 不要生成任何总结性内容和报告

CLAUDE.md

@@ -4,5 +4,5 @@
 - 除非用户明确要求,否则代码中不要添加任何注释
 - 除非用户明确要求,否则不要生成和添加任何文档
 - 不要运行任何测试,不要添加任何测试脚本和代码,不要进行语法检查
-
+- 不要进行任何语法检查
-# 代码说明
+- 不要生成任何总结性内容和报告

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(
+    # query_input = AdaptiveQueryInput(
-        index_name="siem-network-traffic",
+    #     index_name="siem-network-traffic",
-        time_range_start=time_range_start,
+    #     time_range_start=time_range_start,
-        time_range_end=time_range_end,
+    #     time_range_end=time_range_end,
-        filters={
+    #     filters={
-            "event.dataset": "network",
+    #         "event.dataset": "network",
-            "destination.ip": "104.21.11.22",
+    #         "destination.ip": "104.21.11.22",
-            "event.action": "deny"
+    #         "event.action": "deny"
-        }
+    #     }
-    )
+    # )
-
+    #
-    result = toolkit.execute_adaptive_query(query_input)
+    # result = toolkit.execute_adaptive_query(query_input)
-    print(f"Using time field: {query_input.time_field}")
+    # print(f"Using time field: {query_input.time_field}")
-    print(f"Status: {result.status}")
+    # print(f"Status: {result.status}")
-    print(f"Total Hits: {result.total_hits}")
+    # print(f"Total Hits: {result.total_hits}")
-
+    #
-    if result.records:
+    # if result.records:
-        # 验证返回数据
+    #     # 验证返回数据
-        print(f"Sample: {result.records[0]}")
+    #     print(f"Sample: {result.records[0]}")
 
 
 if __name__ == "__main__":

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
+    field_name: str = Field(
-    top_values: Dict[str, int]
+        ...,
+        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")
+    status: str = Field(
-    total_hits: int
+        ...,
-    message: str
+        description=(
-    statistics: List[FieldStat]
+            "Response type indicator based on result volume. "
-    records: List[Dict[str, Any]]
+            "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"
+        )
+    )

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"]:
+                # The aggregation key uses .keyword suffix, but we report original field name
-                    buckets = response["aggregations"][field]["buckets"]
+                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)
     # ==========================================