From 36bee3aedf84fbbfa5463ab820947ae8ae8cea65 Mon Sep 17 00:00:00 2001 From: rookit Date: Thu, 16 Apr 2026 15:31:57 +0800 Subject: [PATCH] tmp --- Lib/basemodule.py | 2 +- ...ivilege-Escalation-via-AttachUserPolicy.py | 447 +++++++----------- ...ege-Escalation-via-AttachUserPolicy_old.py | 310 ++++++++++++ PLUGINS/Redis/redis_stream_api.py | 58 ++- PLUGINS/SIRP/sirpapi.py | 12 +- PLUGINS/SIRP/sirpmodel.py | 308 ++++++------ 6 files changed, 673 insertions(+), 464 deletions(-) create mode 100644 MODULES/Cloud-01-AWS-IAM-Privilege-Escalation-via-AttachUserPolicy_old.py diff --git a/Lib/basemodule.py b/Lib/basemodule.py index 6199a48..38ee676 100644 --- a/Lib/basemodule.py +++ b/Lib/basemodule.py @@ -21,7 +21,7 @@ class BaseModule(BaseAPI): """读取消息""" redis_stream_api = RedisStreamAPI() if self.debug_message_id is not None: - message = redis_stream_api.read_stream_from_start(self.module_name, start_id=self.debug_message_id) + message = redis_stream_api.read_message_by_id(self.module_name, message_id=self.debug_message_id) else: message = redis_stream_api.read_message(stream_key=self.module_name, consumer_group=REDIS_CONSUMER_GROUP, consumer_name=self._thread_name) return message diff --git a/MODULES/Cloud-01-AWS-IAM-Privilege-Escalation-via-AttachUserPolicy.py b/MODULES/Cloud-01-AWS-IAM-Privilege-Escalation-via-AttachUserPolicy.py index 09d17b1..3303718 100644 --- a/MODULES/Cloud-01-AWS-IAM-Privilege-Escalation-via-AttachUserPolicy.py +++ b/MODULES/Cloud-01-AWS-IAM-Privilege-Escalation-via-AttachUserPolicy.py @@ -1,301 +1,204 @@ import json -from datetime import datetime -from typing import Optional, Union, Dict, Any, List - -from langchain_core.messages import HumanMessage -from langgraph.graph import StateGraph, END -from langgraph.graph.state import CompiledStateGraph -from langgraph.types import Command -from pydantic import BaseModel, Field +from typing import List from Lib.api import get_current_time_str -from Lib.basemodule import LanggraphModule -from Lib.llmapi import BaseAgentState -from PLUGINS.LLM.llmapi import LLMAPI +from Lib.basemodule import BaseModule from PLUGINS.SIRP.grouprule import GroupRule, CorrelationConfig from PLUGINS.SIRP.sirpapi import Alert, Case from PLUGINS.SIRP.sirpmodel import AlertModel, ArtifactModel, ArtifactType, ArtifactRole, Severity, AlertStatus, AlertAnalyticType, ProductCategory, Confidence, \ ImpactLevel, AlertRiskLevel, Disposition, AlertAction, AlertPolicyType, CaseModel, CaseStatus, CasePriority -class AnalyzeResult(BaseModel): - """Structure for extracting C2 communication analysis result""" - original_severity: Severity = Field(description="Original alert severity", default=Severity.UNKNOWN) - new_severity: Severity = Field(description="Recommended new severity level", default=Severity.UNKNOWN) - confidence: Confidence = Field(description="Confidence level assessment", default=Confidence.UNKNOWN) - analysis_rationale: str = Field(description="Analysis process and reasons", default=None) - attack_stage: Optional[Union[str, Dict[str, Any]]] = Field(description="e.g., 'T1071 - Application Layer Protocol', 'Command and Control'", default=None) - recommended_actions: Optional[Union[str, Dict[str, Any]]] = Field(description="e.g., 'Isolate host 10.1.1.5'", default=None) - - -class AgentState(BaseAgentState): - analyze_result: AnalyzeResult = None - - -class Module(LanggraphModule): +class Module(BaseModule): THREAD_NUM = 2 def __init__(self): super().__init__() - self.init() - def init(self): - def alert_preprocess_node(state: AgentState): - """ - Read one alert from Redis Stream and preprocess into AlertModel. - Extract artifacts and create AlertModel with proper structure. - """ - raw_message = self.read_message() - if raw_message is None: - return Command(update={}, goto=END) + def run(self): + # 获取原始告警JSON + message = self.read_message() - alert_raw = raw_message + alert_raw = message - event_name = alert_raw.get("eventName", "AttachUserPolicy") - event_time = alert_raw.get("eventTime", alert_raw.get("@timestamp", "")) - rule_name = "AWS IAM Privilege Escalation via AttachUserPolicy" + # 解析需要的字段 + event_name = alert_raw.get("eventName", "AttachUserPolicy") + event_time = alert_raw.get("eventTime", alert_raw.get("@timestamp", "")) + rule_name = "AWS IAM Privilege Escalation via AttachUserPolicy" - user_identity = alert_raw.get("userIdentity", {}) - principal_user = user_identity.get("userName", "unknown-user") - principal_arn = user_identity.get("arn", "unknown-arn") - principal_id = user_identity.get("principalId", "unknown-id") - access_key_id = user_identity.get("accessKeyId", "unknown-key") + user_identity = alert_raw.get("userIdentity", {}) + principal_user = user_identity.get("userName", "unknown-user") + principal_arn = user_identity.get("arn", "unknown-arn") + principal_id = user_identity.get("principalId", "unknown-id") + access_key_id = user_identity.get("accessKeyId", "unknown-key") - request_params = alert_raw.get("requestParameters", {}) - target_user = request_params.get("userName", "unknown-target") - policy_arn = request_params.get("policyArn", "unknown-policy") + request_params = alert_raw.get("requestParameters", {}) + target_user = request_params.get("userName", "unknown-target") + policy_arn = request_params.get("policyArn", "unknown-policy") - source_ip = alert_raw.get("sourceIPAddress", "unknown-ip") - aws_region = alert_raw.get("awsRegion", "unknown-region") - account_id = alert_raw.get("recipientAccountId", alert_raw.get("cloud.account.id", "unknown-account")) - user_agent = alert_raw.get("userAgent", "unknown-agent") - event_id = alert_raw.get("eventID", "") + source_ip = alert_raw.get("sourceIPAddress", "unknown-ip") + aws_region = alert_raw.get("awsRegion", "unknown-region") + account_id = alert_raw.get("recipientAccountId", alert_raw.get("cloud.account.id", "unknown-account")) + user_agent = alert_raw.get("userAgent", "unknown-agent") + event_id = alert_raw.get("eventID", "") - risk_score = alert_raw.get("event.risk_score", alert_raw.get("risk_score", 100)) - log_level = alert_raw.get("log.level", "critical") - message = alert_raw.get("message", "") + risk_score = alert_raw.get("event.risk_score", alert_raw.get("risk_score", 100)) + log_level = alert_raw.get("log.level", "critical") + message = alert_raw.get("message", "") - severity_map = { - "critical": Severity.CRITICAL, - "high": Severity.HIGH, - "medium": Severity.MEDIUM, - "low": Severity.LOW, - "informational": Severity.INFORMATIONAL, - } - severity = severity_map.get(log_level.lower(), Severity.CRITICAL) + severity_map = { + "critical": Severity.CRITICAL, + "high": Severity.HIGH, + "medium": Severity.MEDIUM, + "low": Severity.LOW, + "informational": Severity.INFORMATIONAL, + } + severity = severity_map.get(log_level.lower(), Severity.CRITICAL) - event_time_formatted = event_time if event_time else get_current_time_str() + event_time_formatted = event_time if event_time else get_current_time_str() - artifacts: List[ArtifactModel] = [ - ArtifactModel( - type=ArtifactType.USER, - role=ArtifactRole.ACTOR, - value=principal_user, - name="Principal User" - ), ArtifactModel( - type=ArtifactType.OTHER, - role=ArtifactRole.ACTOR, - value=principal_arn, - name="Principal ARN" - ), ArtifactModel( - type=ArtifactType.USER, - role=ArtifactRole.TARGET, - value=target_user, - name="Target User" - ), ArtifactModel( - type=ArtifactType.IP_ADDRESS, - role=ArtifactRole.ACTOR, - value=source_ip, - name="Source IP" - ), ArtifactModel( - type=ArtifactType.OTHER, - role=ArtifactRole.RELATED, - value=policy_arn, - name="Policy ARN" - ), ArtifactModel( - type=ArtifactType.OTHER, - role=ArtifactRole.RELATED, - value=account_id, - name="AWS Account ID" + # 提取 artifact + + artifacts: List[ArtifactModel] = [ + ArtifactModel( + type=ArtifactType.USER, + role=ArtifactRole.ACTOR, + value=principal_user, + name="Principal User" + ), ArtifactModel( + type=ArtifactType.OTHER, + role=ArtifactRole.ACTOR, + value=principal_arn, + name="Principal ARN" + ), ArtifactModel( + type=ArtifactType.USER, + role=ArtifactRole.TARGET, + value=target_user, + name="Target User" + ), ArtifactModel( + type=ArtifactType.IP_ADDRESS, + role=ArtifactRole.ACTOR, + value=source_ip, + name="Source IP" + ), ArtifactModel( + type=ArtifactType.OTHER, + role=ArtifactRole.RELATED, + value=policy_arn, + name="Policy ARN" + ), ArtifactModel( + type=ArtifactType.OTHER, + role=ArtifactRole.RELATED, + value=account_id, + name="AWS Account ID" + ) + ] + + if access_key_id and access_key_id != "unknown-key": + artifacts.append(ArtifactModel( + type=ArtifactType.OTHER, + role=ArtifactRole.ACTOR, + value=access_key_id, + name="Access Key ID" + )) + + # 计算 correlation + correlation_config = CorrelationConfig( + rule_id=self.module_name, + time_window="24h", + keys=[principal_user, target_user, account_id] + ) + group_rule = GroupRule(config=correlation_config) + correlation_uid = group_rule.generate_correlation_uid(timestamp=event_time_formatted) + + # 拼装 Alert + alert_model = AlertModel( + title=f"AWS IAM Privilege Escalation: {principal_user} attached {policy_arn.split('/')[-1]} to {target_user}", + src_url=f"AWS CloudTrail - Event ID: {event_id}", + severity=severity, + status=AlertStatus.NEW, + status_detail="New alert received from AWS CloudTrail - awaiting analysis", + disposition=Disposition.DETECTED, + action=AlertAction.OBSERVED, + rule_id=self.module_name, + rule_name=rule_name, + source_uid=event_id, + correlation_uid=correlation_uid, + count=1, + analytic_type=AlertAnalyticType.BEHAVIORAL, + analytic_name="AWS IAM Behavioral Anomaly Detection", + analytic_desc="Detects suspicious IAM policy attachment operations that indicate privilege escalation attempts or backdoor account creation", + analytic_state=None, + product_category=ProductCategory.CLOUD, + product_name="AWS CloudTrail", + product_vendor="Amazon AWS", + product_feature="CloudTrail Logging", + first_seen_time=event_time_formatted, + last_seen_time=event_time_formatted, + desc=message or f"IAM user {principal_user} attached policy {policy_arn} to user {target_user} in account {account_id}", + data_sources=["AWS CloudTrail"], + labels=["iam-privilege-escalation", "aws-cloudtrail", f"account-{account_id}"], + raw_data=json.dumps(alert_raw), + unmapped=json.dumps({ + "userAgent": user_agent, + "awsRegion": aws_region, + "requestID": alert_raw.get("requestID", ""), + "eventVersion": alert_raw.get("eventVersion", "") + }), + tactic="T1098.003 - AWS IAM Account Manipulation", + technique="Privilege Escalation", + sub_technique="Create IAM Access Keys", + mitigation="Enable IAM access analyzer, enforce MFA, use service control policies to restrict policy attachment", + policy_name="AWS IAM Access Policy", + policy_type=AlertPolicyType.IDENTITY_POLICY, + policy_desc="IAM identity-based policy that grants permissions to AWS services", + impact=ImpactLevel.CRITICAL if severity in [Severity.CRITICAL, Severity.HIGH] else ImpactLevel.HIGH, + confidence=Confidence.HIGH, + risk_level=AlertRiskLevel.CRITICAL if severity == Severity.CRITICAL else AlertRiskLevel.HIGH, + risk_details=f"Unauthorized administrator policy attached to {target_user} - potential backdoor account creation or privilege escalation attack" + ) + + alert_model.artifacts = artifacts + + saved_alert_rowid = Alert.create(alert_model) + alert_model.rowid = saved_alert_rowid + + self.logger.debug(f"Alert created with Rowid: {saved_alert_rowid}") + + try: + existing_case = Case.get_by_correlation_uid(correlation_uid, lazy_load=True) + + if existing_case is not None: + self.logger.debug(f"Found existing case with correlation_uid: {correlation_uid}, Case ID: {existing_case.rowid}") + + # 将alert 挂载到已有 case , 也可以根据需求更新 case 其他字段 + update_case = CaseModel(alerts=[*existing_case.alerts, saved_alert_rowid], rowid=existing_case.rowid) + Case.update(update_case) + + self.logger.debug(f"Alert {saved_alert_rowid} added to existing case {existing_case.rowid}") + + else: + self.logger.debug(f"No existing case found for correlation_uid: {correlation_uid}, creating new case") + + new_case = CaseModel( + title=f"AWS IAM Privilege Escalation: {principal_user} → {target_user}", + severity=severity, + impact=ImpactLevel.CRITICAL if severity in [Severity.CRITICAL, Severity.HIGH] else ImpactLevel.HIGH, + priority=CasePriority.CRITICAL if severity == Severity.CRITICAL else CasePriority.HIGH, + confidence=Confidence.HIGH, + status=CaseStatus.NEW, + description=f"AWS IAM privilege escalation detected: {principal_user} attached {policy_arn} to {target_user}", + category=ProductCategory.CLOUD, + tags=["iam-privilege-escalation", "aws-cloudtrail", f"account-{account_id}"], + correlation_uid=correlation_uid, + alerts=[saved_alert_rowid] ) - ] - if access_key_id and access_key_id != "unknown-key": - artifacts.append(ArtifactModel( - type=ArtifactType.OTHER, - role=ArtifactRole.ACTOR, - value=access_key_id, - name="Access Key ID" - )) + case_uid = Case.create(new_case) + self.logger.debug(f"New case created with UID: {case_uid}, Alert: {saved_alert_rowid}") - correlation_config = CorrelationConfig( - rule_id=self.module_name, - time_window="24h", - keys=[principal_user, target_user, account_id] - ) - group_rule = GroupRule(config=correlation_config) - correlation_uid = group_rule.generate_correlation_uid(timestamp=event_time_formatted) + except Exception as e: + self.logger.error(f"Error creating/updating case for correlation_uid {correlation_uid}: {str(e)}") - alert_model = AlertModel( - title=f"AWS IAM Privilege Escalation: {principal_user} attached {policy_arn.split('/')[-1]} to {target_user}", - src_url=f"AWS CloudTrail - Event ID: {event_id}", - severity=severity, - status=AlertStatus.NEW, - status_detail="New alert received from AWS CloudTrail - awaiting analysis", - disposition=Disposition.DETECTED, - action=AlertAction.OBSERVED, - rule_id=self.module_name, - rule_name=rule_name, - source_uid=event_id, - correlation_uid=correlation_uid, - count=1, - analytic_type=AlertAnalyticType.BEHAVIORAL, - analytic_name="AWS IAM Behavioral Anomaly Detection", - analytic_desc="Detects suspicious IAM policy attachment operations that indicate privilege escalation attempts or backdoor account creation", - analytic_state=None, - product_category=ProductCategory.CLOUD, - product_name="AWS CloudTrail", - product_vendor="Amazon AWS", - product_feature="CloudTrail Logging", - first_seen_time=event_time_formatted, - last_seen_time=event_time_formatted, - desc=message or f"IAM user {principal_user} attached policy {policy_arn} to user {target_user} in account {account_id}", - data_sources=["AWS CloudTrail"], - labels=["iam-privilege-escalation", "aws-cloudtrail", f"account-{account_id}"], - raw_data=json.dumps(alert_raw), - unmapped=json.dumps({ - "userAgent": user_agent, - "awsRegion": aws_region, - "requestID": alert_raw.get("requestID", ""), - "eventVersion": alert_raw.get("eventVersion", "") - }), - tactic="T1098.003 - AWS IAM Account Manipulation", - technique="Privilege Escalation", - sub_technique="Create IAM Access Keys", - mitigation="Enable IAM access analyzer, enforce MFA, use service control policies to restrict policy attachment", - policy_name="AWS IAM Access Policy", - policy_type=AlertPolicyType.IDENTITY_POLICY, - policy_desc="IAM identity-based policy that grants permissions to AWS services", - impact=ImpactLevel.CRITICAL if severity in [Severity.CRITICAL, Severity.HIGH] else ImpactLevel.HIGH, - confidence=Confidence.HIGH, - risk_level=AlertRiskLevel.CRITICAL if severity == Severity.CRITICAL else AlertRiskLevel.HIGH, - risk_details=f"Unauthorized administrator policy attached to {target_user} - potential backdoor account creation or privilege escalation attack" - ) - - alert_model.artifacts = artifacts - - saved_alert_rowid = Alert.create(alert_model) - alert_model.rowid = saved_alert_rowid - - self.logger.debug(f"Alert created with Rowid: {saved_alert_rowid}") - - try: - existing_cases = Case.list_by_correlation_uid(correlation_uid, lazy_load=True) - - if existing_cases and len(existing_cases) > 0: - existing_case = existing_cases[0] - self.logger.debug(f"Found existing case with correlation_uid: {correlation_uid}, Case ID: {existing_case.rowid}") - - update_case = CaseModel(alerts=[*existing_case.alerts, saved_alert_rowid], rowid=existing_case.rowid) - Case.update(update_case) - - self.logger.debug(f"Alert {saved_alert_rowid} added to existing case {existing_case.rowid}") - - else: - self.logger.debug(f"No existing case found for correlation_uid: {correlation_uid}, creating new case") - - new_case = CaseModel( - title=f"AWS IAM Privilege Escalation: {principal_user} → {target_user}", - severity=severity, - impact=ImpactLevel.CRITICAL if severity in [Severity.CRITICAL, Severity.HIGH] else ImpactLevel.HIGH, - priority=CasePriority.CRITICAL if severity == Severity.CRITICAL else CasePriority.HIGH, - confidence=Confidence.HIGH, - status=CaseStatus.NEW, - description=f"AWS IAM privilege escalation detected: {principal_user} attached {policy_arn} to {target_user}", - category=ProductCategory.CLOUD, - tags=["iam-privilege-escalation", "aws-cloudtrail", f"account-{account_id}"], - correlation_uid=correlation_uid, - alerts=[saved_alert_rowid] - ) - - case_uid = Case.create(new_case) - self.logger.debug(f"New case created with UID: {case_uid}, Alert: {saved_alert_rowid}") - - except Exception as e: - self.logger.error(f"Error creating/updating case for correlation_uid {correlation_uid}: {str(e)}") - # return Command(update={}, goto=END) - return {"alert": alert_model} - - def alert_analyze_node(state: AgentState): - """ - Analyze the alert using AI (LLM) with structured few-shot examples for AWS IAM privilege escalation detection. - Leverages threat intelligence and cloud behavior patterns. - """ - system_prompt_template = self.load_system_prompt_template("senior_cloud_security_expert") - - current_date = datetime.now().strftime("%Y-%m-%d") - system_message = system_prompt_template.format(current_date=current_date) - - few_shot_examples = [ - ] - - alert = state.alert - messages = [ - system_message, - *few_shot_examples, - HumanMessage(content=alert.model_dump_json_for_ai()), - ] - - llm_api = LLMAPI() - llm = llm_api.get_model(tag=["fast", "structured_output"]) - llm_structured = llm.with_structured_output(AnalyzeResult) - response: AnalyzeResult = llm_structured.invoke(messages) - - return {"analyze_result": response} - - def alert_output_node(state: AgentState): - """ - Save analysis result to AlertModel and persist using SIRP API. - Updates severity, confidence, and enriches with AI-generated insights. - """ - alert_model: AlertModel = state.alert - analyze_result: AnalyzeResult = state.analyze_result - - alert_model.severity = analyze_result.new_severity - alert_model.confidence = analyze_result.confidence - alert_model.summary_ai = str(analyze_result.analysis_rationale) - - if analyze_result.recommended_actions: - alert_model.remediation = str(analyze_result.recommended_actions) - - labels = list(alert_model.labels) if alert_model.labels else [] - if analyze_result.attack_stage and analyze_result.confidence in [Confidence.HIGH, Confidence.MEDIUM]: - labels.append("confirmed-privilege-escalation") - if analyze_result.new_severity in [Severity.CRITICAL, Severity.HIGH]: - labels.append("high-priority-incident") - alert_model.labels = labels - - alert_model.uid = f"iam-priv-esc-{get_current_time_str()}" - - update_alert_rowid = Alert.update(alert_model) - - self.logger.info( - f"AWS IAM Privilege Escalation Alert saved with RowID: {update_alert_rowid}, Severity: {analyze_result.new_severity}, Confidence: {analyze_result.confidence}") - - return {} - - workflow = StateGraph(AgentState) - - workflow.add_node("alert_preprocess_node", alert_preprocess_node) - workflow.add_node("alert_analyze_node", alert_analyze_node) - workflow.add_node("alert_output_node", alert_output_node) - - workflow.set_entry_point("alert_preprocess_node") - workflow.add_edge("alert_preprocess_node", "alert_analyze_node") - workflow.add_edge("alert_analyze_node", "alert_output_node") - workflow.set_finish_point("alert_output_node") - - self.graph: CompiledStateGraph = workflow.compile(checkpointer=self.get_checkpointer()) return True @@ -306,5 +209,5 @@ if __name__ == "__main__": os.environ.setdefault("DJANGO_SETTINGS_MODULE", "ASP.settings") django.setup() module = Module() - module.debug_message_id = "1772179651858-0" + module.debug_message_id = "1776307910636-0" module.run() diff --git a/MODULES/Cloud-01-AWS-IAM-Privilege-Escalation-via-AttachUserPolicy_old.py b/MODULES/Cloud-01-AWS-IAM-Privilege-Escalation-via-AttachUserPolicy_old.py new file mode 100644 index 0000000..7bebb5c --- /dev/null +++ b/MODULES/Cloud-01-AWS-IAM-Privilege-Escalation-via-AttachUserPolicy_old.py @@ -0,0 +1,310 @@ +import json +from datetime import datetime +from typing import Optional, Union, Dict, Any, List + +from langchain_core.messages import HumanMessage +from langgraph.graph import StateGraph, END +from langgraph.graph.state import CompiledStateGraph +from langgraph.types import Command +from pydantic import BaseModel, Field + +from Lib.api import get_current_time_str +from Lib.basemodule import LanggraphModule +from Lib.llmapi import BaseAgentState +from PLUGINS.LLM.llmapi import LLMAPI +from PLUGINS.SIRP.grouprule import GroupRule, CorrelationConfig +from PLUGINS.SIRP.sirpapi import Alert, Case +from PLUGINS.SIRP.sirpmodel import AlertModel, ArtifactModel, ArtifactType, ArtifactRole, Severity, AlertStatus, AlertAnalyticType, ProductCategory, Confidence, \ + ImpactLevel, AlertRiskLevel, Disposition, AlertAction, AlertPolicyType, CaseModel, CaseStatus, CasePriority + + +class AnalyzeResult(BaseModel): + """Structure for extracting C2 communication analysis result""" + original_severity: Severity = Field(description="Original alert severity", default=Severity.UNKNOWN) + new_severity: Severity = Field(description="Recommended new severity level", default=Severity.UNKNOWN) + confidence: Confidence = Field(description="Confidence level assessment", default=Confidence.UNKNOWN) + analysis_rationale: str = Field(description="Analysis process and reasons", default=None) + attack_stage: Optional[Union[str, Dict[str, Any]]] = Field(description="e.g., 'T1071 - Application Layer Protocol', 'Command and Control'", default=None) + recommended_actions: Optional[Union[str, Dict[str, Any]]] = Field(description="e.g., 'Isolate host 10.1.1.5'", default=None) + + +class AgentState(BaseAgentState): + analyze_result: AnalyzeResult = None + + +class Module(LanggraphModule): + THREAD_NUM = 2 + + def __init__(self): + super().__init__() + self.init() + + def init(self): + def alert_preprocess_node(state: AgentState): + """ + Read one alert from Redis Stream and preprocess into AlertModel. + Extract artifacts and create AlertModel with proper structure. + """ + raw_message = self.read_message() + if raw_message is None: + return Command(update={}, goto=END) + + alert_raw = raw_message + + event_name = alert_raw.get("eventName", "AttachUserPolicy") + event_time = alert_raw.get("eventTime", alert_raw.get("@timestamp", "")) + rule_name = "AWS IAM Privilege Escalation via AttachUserPolicy" + + user_identity = alert_raw.get("userIdentity", {}) + principal_user = user_identity.get("userName", "unknown-user") + principal_arn = user_identity.get("arn", "unknown-arn") + principal_id = user_identity.get("principalId", "unknown-id") + access_key_id = user_identity.get("accessKeyId", "unknown-key") + + request_params = alert_raw.get("requestParameters", {}) + target_user = request_params.get("userName", "unknown-target") + policy_arn = request_params.get("policyArn", "unknown-policy") + + source_ip = alert_raw.get("sourceIPAddress", "unknown-ip") + aws_region = alert_raw.get("awsRegion", "unknown-region") + account_id = alert_raw.get("recipientAccountId", alert_raw.get("cloud.account.id", "unknown-account")) + user_agent = alert_raw.get("userAgent", "unknown-agent") + event_id = alert_raw.get("eventID", "") + + risk_score = alert_raw.get("event.risk_score", alert_raw.get("risk_score", 100)) + log_level = alert_raw.get("log.level", "critical") + message = alert_raw.get("message", "") + + severity_map = { + "critical": Severity.CRITICAL, + "high": Severity.HIGH, + "medium": Severity.MEDIUM, + "low": Severity.LOW, + "informational": Severity.INFORMATIONAL, + } + severity = severity_map.get(log_level.lower(), Severity.CRITICAL) + + event_time_formatted = event_time if event_time else get_current_time_str() + + artifacts: List[ArtifactModel] = [ + ArtifactModel( + type=ArtifactType.USER, + role=ArtifactRole.ACTOR, + value=principal_user, + name="Principal User" + ), ArtifactModel( + type=ArtifactType.OTHER, + role=ArtifactRole.ACTOR, + value=principal_arn, + name="Principal ARN" + ), ArtifactModel( + type=ArtifactType.USER, + role=ArtifactRole.TARGET, + value=target_user, + name="Target User" + ), ArtifactModel( + type=ArtifactType.IP_ADDRESS, + role=ArtifactRole.ACTOR, + value=source_ip, + name="Source IP" + ), ArtifactModel( + type=ArtifactType.OTHER, + role=ArtifactRole.RELATED, + value=policy_arn, + name="Policy ARN" + ), ArtifactModel( + type=ArtifactType.OTHER, + role=ArtifactRole.RELATED, + value=account_id, + name="AWS Account ID" + ) + ] + + if access_key_id and access_key_id != "unknown-key": + artifacts.append(ArtifactModel( + type=ArtifactType.OTHER, + role=ArtifactRole.ACTOR, + value=access_key_id, + name="Access Key ID" + )) + + correlation_config = CorrelationConfig( + rule_id=self.module_name, + time_window="24h", + keys=[principal_user, target_user, account_id] + ) + group_rule = GroupRule(config=correlation_config) + correlation_uid = group_rule.generate_correlation_uid(timestamp=event_time_formatted) + + alert_model = AlertModel( + title=f"AWS IAM Privilege Escalation: {principal_user} attached {policy_arn.split('/')[-1]} to {target_user}", + src_url=f"AWS CloudTrail - Event ID: {event_id}", + severity=severity, + status=AlertStatus.NEW, + status_detail="New alert received from AWS CloudTrail - awaiting analysis", + disposition=Disposition.DETECTED, + action=AlertAction.OBSERVED, + rule_id=self.module_name, + rule_name=rule_name, + source_uid=event_id, + correlation_uid=correlation_uid, + count=1, + analytic_type=AlertAnalyticType.BEHAVIORAL, + analytic_name="AWS IAM Behavioral Anomaly Detection", + analytic_desc="Detects suspicious IAM policy attachment operations that indicate privilege escalation attempts or backdoor account creation", + analytic_state=None, + product_category=ProductCategory.CLOUD, + product_name="AWS CloudTrail", + product_vendor="Amazon AWS", + product_feature="CloudTrail Logging", + first_seen_time=event_time_formatted, + last_seen_time=event_time_formatted, + desc=message or f"IAM user {principal_user} attached policy {policy_arn} to user {target_user} in account {account_id}", + data_sources=["AWS CloudTrail"], + labels=["iam-privilege-escalation", "aws-cloudtrail", f"account-{account_id}"], + raw_data=json.dumps(alert_raw), + unmapped=json.dumps({ + "userAgent": user_agent, + "awsRegion": aws_region, + "requestID": alert_raw.get("requestID", ""), + "eventVersion": alert_raw.get("eventVersion", "") + }), + tactic="T1098.003 - AWS IAM Account Manipulation", + technique="Privilege Escalation", + sub_technique="Create IAM Access Keys", + mitigation="Enable IAM access analyzer, enforce MFA, use service control policies to restrict policy attachment", + policy_name="AWS IAM Access Policy", + policy_type=AlertPolicyType.IDENTITY_POLICY, + policy_desc="IAM identity-based policy that grants permissions to AWS services", + impact=ImpactLevel.CRITICAL if severity in [Severity.CRITICAL, Severity.HIGH] else ImpactLevel.HIGH, + confidence=Confidence.HIGH, + risk_level=AlertRiskLevel.CRITICAL if severity == Severity.CRITICAL else AlertRiskLevel.HIGH, + risk_details=f"Unauthorized administrator policy attached to {target_user} - potential backdoor account creation or privilege escalation attack" + ) + + alert_model.artifacts = artifacts + + saved_alert_rowid = Alert.create(alert_model) + alert_model.rowid = saved_alert_rowid + + self.logger.debug(f"Alert created with Rowid: {saved_alert_rowid}") + + try: + existing_cases = Case.get_by_correlation_uid(correlation_uid, lazy_load=True) + + if existing_cases and len(existing_cases) > 0: + existing_case = existing_cases[0] + self.logger.debug(f"Found existing case with correlation_uid: {correlation_uid}, Case ID: {existing_case.rowid}") + + update_case = CaseModel(alerts=[*existing_case.alerts, saved_alert_rowid], rowid=existing_case.rowid) + Case.update(update_case) + + self.logger.debug(f"Alert {saved_alert_rowid} added to existing case {existing_case.rowid}") + + else: + self.logger.debug(f"No existing case found for correlation_uid: {correlation_uid}, creating new case") + + new_case = CaseModel( + title=f"AWS IAM Privilege Escalation: {principal_user} → {target_user}", + severity=severity, + impact=ImpactLevel.CRITICAL if severity in [Severity.CRITICAL, Severity.HIGH] else ImpactLevel.HIGH, + priority=CasePriority.CRITICAL if severity == Severity.CRITICAL else CasePriority.HIGH, + confidence=Confidence.HIGH, + status=CaseStatus.NEW, + description=f"AWS IAM privilege escalation detected: {principal_user} attached {policy_arn} to {target_user}", + category=ProductCategory.CLOUD, + tags=["iam-privilege-escalation", "aws-cloudtrail", f"account-{account_id}"], + correlation_uid=correlation_uid, + alerts=[saved_alert_rowid] + ) + + case_uid = Case.create(new_case) + self.logger.debug(f"New case created with UID: {case_uid}, Alert: {saved_alert_rowid}") + + except Exception as e: + self.logger.error(f"Error creating/updating case for correlation_uid {correlation_uid}: {str(e)}") + # return Command(update={}, goto=END) + return {"alert": alert_model} + + def alert_analyze_node(state: AgentState): + """ + Analyze the alert using AI (LLM) with structured few-shot examples for AWS IAM privilege escalation detection. + Leverages threat intelligence and cloud behavior patterns. + """ + system_prompt_template = self.load_system_prompt_template("senior_cloud_security_expert") + + current_date = datetime.now().strftime("%Y-%m-%d") + system_message = system_prompt_template.format(current_date=current_date) + + few_shot_examples = [ + ] + + alert = state.alert + messages = [ + system_message, + *few_shot_examples, + HumanMessage(content=alert.model_dump_json_for_ai()), + ] + + llm_api = LLMAPI() + llm = llm_api.get_model(tag=["fast", "structured_output"]) + llm_structured = llm.with_structured_output(AnalyzeResult) + response: AnalyzeResult = llm_structured.invoke(messages) + + return {"analyze_result": response} + + def alert_output_node(state: AgentState): + """ + Save analysis result to AlertModel and persist using SIRP API. + Updates severity, confidence, and enriches with AI-generated insights. + """ + alert_model: AlertModel = state.alert + analyze_result: AnalyzeResult = state.analyze_result + + alert_model.severity = analyze_result.new_severity + alert_model.confidence = analyze_result.confidence + alert_model.summary_ai = str(analyze_result.analysis_rationale) + + if analyze_result.recommended_actions: + alert_model.remediation = str(analyze_result.recommended_actions) + + labels = list(alert_model.labels) if alert_model.labels else [] + if analyze_result.attack_stage and analyze_result.confidence in [Confidence.HIGH, Confidence.MEDIUM]: + labels.append("confirmed-privilege-escalation") + if analyze_result.new_severity in [Severity.CRITICAL, Severity.HIGH]: + labels.append("high-priority-incident") + alert_model.labels = labels + + alert_model.uid = f"iam-priv-esc-{get_current_time_str()}" + + update_alert_rowid = Alert.update(alert_model) + + self.logger.info( + f"AWS IAM Privilege Escalation Alert saved with RowID: {update_alert_rowid}, Severity: {analyze_result.new_severity}, Confidence: {analyze_result.confidence}") + + return {} + + workflow = StateGraph(AgentState) + + workflow.add_node("alert_preprocess_node", alert_preprocess_node) + workflow.add_node("alert_analyze_node", alert_analyze_node) + workflow.add_node("alert_output_node", alert_output_node) + + workflow.set_entry_point("alert_preprocess_node") + workflow.add_edge("alert_preprocess_node", "alert_analyze_node") + workflow.add_edge("alert_analyze_node", "alert_output_node") + workflow.set_finish_point("alert_output_node") + + self.graph: CompiledStateGraph = workflow.compile(checkpointer=self.get_checkpointer()) + return True + + +if __name__ == "__main__": + import os + import django + + os.environ.setdefault("DJANGO_SETTINGS_MODULE", "ASP.settings") + django.setup() + module = Module() + module.debug_message_id = "1772179651858-0" + module.run() diff --git a/PLUGINS/Redis/redis_stream_api.py b/PLUGINS/Redis/redis_stream_api.py index 12cbff4..141cb9d 100644 --- a/PLUGINS/Redis/redis_stream_api.py +++ b/PLUGINS/Redis/redis_stream_api.py @@ -1,5 +1,6 @@ import json import time +from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError from typing import Dict, Any, Optional, List import redis @@ -106,34 +107,51 @@ class RedisStreamAPI(object): logger.error(f"Error reading from stream {stream_key}: {e}") time.sleep(1) # 发生异常(如网络闪断)时稍作停顿，防止死循环刷屏 - def read_stream_from_start(self, stream_key, start_id='0-0'): + def read_stream_head(self, stream_key: str, n: int, timeout: Optional[float] = None) -> List[Dict[str, Any]]: """ - 从指定位置一次性读取一条消息（非阻塞）. + 读取指定 stream 的前 n 条消息（非阻塞）. :param stream_key: Stream 的名称. - :param start_id: 开始读取的消息ID，默认从头开始. + :param n: 要读取的消息数量. + :param timeout: 超时时间（秒），None 表示不限制. """ + def _fetch(): + messages = self.redis_client.xrange(stream_key, min='-', max='+', count=n) + return [json.loads(fields["data"]) for _, fields in messages] try: - messages = self.redis_client.xread( - count=1, - block=None, - streams={stream_key: start_id} - ) + if timeout is None: + return _fetch() + with ThreadPoolExecutor(max_workers=1) as executor: + return executor.submit(_fetch).result(timeout=timeout) + except FuturesTimeoutError: + logger.error(f"Timeout reading stream head: {stream_key}") + return [] + except Exception as e: + logger.exception(e) + return [] - if not messages or not messages[0][1]: + def read_message_by_id(self, stream_key: str, message_id: str, timeout: Optional[float] = None) -> Optional[Dict[str, Any]]: + """ + 读取指定 ID 的消息（精确匹配，非阻塞）. + :param stream_key: Stream 的名称. + :param message_id: 要读取的消息 ID. + :param timeout: 超时时间（秒），None 表示不限制. + """ + def _fetch(): + messages = self.redis_client.xrange(stream_key, min=message_id, max=message_id, count=1) + if not messages: return None + _, fields = messages[0] + return json.loads(fields["data"]) - # 解析消息 - stream_name, stream_messages = messages[0] - if not stream_messages: - return None - - message_id, fields = stream_messages[0] - - value = fields["data"] - data = json.loads(value) - return data - + try: + if timeout is None: + return _fetch() + with ThreadPoolExecutor(max_workers=1) as executor: + return executor.submit(_fetch).result(timeout=timeout) + except FuturesTimeoutError: + logger.error(f"Timeout reading message {message_id} from: {stream_key}") + return None except Exception as e: logger.exception(e) return None diff --git a/PLUGINS/SIRP/sirpapi.py b/PLUGINS/SIRP/sirpapi.py index d1a32d7..4c4fbc2 100644 --- a/PLUGINS/SIRP/sirpapi.py +++ b/PLUGINS/SIRP/sirpapi.py @@ -314,7 +314,7 @@ class Case(BaseWorksheetEntity[CaseModel]): return model @classmethod - def list_by_correlation_uid(cls, correlation_uid, lazy_load=False) -> List[CaseModel]: + def get_by_correlation_uid(cls, correlation_uid, lazy_load=False) -> Union[CaseModel, None]: """根据correlation_uid查询关联的Case""" filter_model = Group( logic="AND", @@ -326,7 +326,15 @@ class Case(BaseWorksheetEntity[CaseModel]): ) ] ) - return cls.list(filter_model, lazy_load=lazy_load) + cases = cls.list(filter_model, lazy_load=lazy_load) + if len(cases) == 0: + return None + elif len(cases) == 1: + return cases[0] + elif len(cases) > 1: + logger.warning(f"More than one case has correlation_uid : {correlation_uid}") + return cases[0] + return None @classmethod def get_by_id(cls, case_id, lazy_load=False) -> Union[CaseModel, None]: diff --git a/PLUGINS/SIRP/sirpmodel.py b/PLUGINS/SIRP/sirpmodel.py index 8fdbfdd..ff27e92 100644 --- a/PLUGINS/SIRP/sirpmodel.py +++ b/PLUGINS/SIRP/sirpmodel.py @@ -10,16 +10,6 @@ from pydantic import BaseModel, Field, field_validator, ConfigDict, field_serial from PLUGINS.SIRP.nocolymodel import AttachmentModel, AccountModel, AttachmentCreateModel -# region Enums - -# class WfStatus(StrEnum): -# PASS = "通过" -# REJECT = "否决" -# ABORT = "中止" -# IN_PROGRESS = "进行中" -# EMPTY = "" - - class MessageType(StrEnum): SYSTEM = "SystemMessage" HUMAN = "HumanMessage" @@ -291,13 +281,11 @@ class CasePriority(StrEnum): class CaseStatus(StrEnum): - # UNKNOWN = "Unknown" NEW = "New" IN_PROGRESS = "In Progress" ON_HOLD = "On Hold" RESOLVED = "Resolved" CLOSED = "Closed" - # OTHER = "Other" class CaseVerdict(StrEnum): @@ -328,31 +316,17 @@ class KnowledgeAction(StrEnum): DONE = 'Done' -# endregion Enums - - class BaseSystemModel(BaseModel): model_config = ConfigDict(populate_by_name=True) ai_exclude_fields: ClassVar[set[str]] = set() - rowid: Optional[str] = Field(default=None, description="Unique row ID") - ownerid: Optional[AccountModel] = Field(default=None, description="Record owner") - caid: Optional[AccountModel] = Field(default=None, description="Creator account") - ctime: Optional[Union[datetime, str]] = Field(default=None, description="Record created time") - utime: Optional[Union[datetime, str]] = Field(default=None, description="Record last updated time") - uaid: Optional[AccountModel] = Field(default=None, description="Last updated by") - - # 流程相关参数 - # wfname: Optional[str] = Field(default=None, description="Workflow name") - # wfcuaids: Optional[Any] = Field(default=None, description="Current assignees") - # wfcaid: Optional[Any] = Field(default=None, description="Current assignee") - # wfctime: Optional[Union[datetime, str]] = Field(default=None, description="Workflow created at") - # wfrtime: Optional[Union[datetime, str]] = Field(default=None, description="Workflow received at") - # wfcotime: Optional[Union[datetime, str]] = Field(default=None, description="Workflow completed at") - # wfdtime: Optional[Union[datetime, str]] = Field(default=None, description="Workflow due at") - # wfftime: Optional[Any] = Field(default=None, description="Workflow followed at") - # wfstatus: Optional[WfStatus] = Field(default=None, description="Workflow status") + rowid: Optional[str] = Field(default=None, description="唯一行 ID (Unique row ID)") + ownerid: Optional[AccountModel] = Field(default=None, description="记录所有者 (Record owner)") + caid: Optional[AccountModel] = Field(default=None, description="创建账号 (Creator account)") + ctime: Optional[Union[datetime, str]] = Field(default=None, description="记录创建时间 (Record created time)") + utime: Optional[Union[datetime, str]] = Field(default=None, description="记录最后更新时间 (Record last updated time)") + uaid: Optional[AccountModel] = Field(default=None, description="最后更新人 (Last updated by)") @field_validator("ownerid", mode="before") def empty_list_to_none(cls, v): @@ -536,170 +510,166 @@ class BaseSystemModel(BaseModel): class MessageModel(BaseSystemModel): - playbook: Optional[List[Union[PlaybookModel, str]]] = Field(default="", description="Owning playbook row ID") - node: Optional[str] = Field(default="", description="Source node name or ID") - content: Optional[str] = Field(default="", description="Message text content") - data: Optional[str] = Field(default="", description="Message JSON payload") + playbook: Optional[List[Union[PlaybookModel, str]]] = Field(default="", description="所属剧本行 ID (Owning playbook row ID)") + node: Optional[str] = Field(default="", description="源节点名称或 ID (Source node name or ID)") + content: Optional[str] = Field(default="", description="消息文本内容 (Message text content)") + data: Optional[str] = Field(default="", description="消息 JSON 负载 (Message JSON payload)") type: Optional[MessageType] = Field(default=None, - description="Message role type") + description="消息角色类型 (Message role type)") class PlaybookModel(BaseSystemModel): - id: Optional[str] = Field(default=None) - source_rowid: Optional[str] = Field(default="", description="Trigger source row ID") - source_id: Optional[str] = Field(default="", description="Trigger source record ID e.g. case_00000_1,alert_000001,artifact_000001") - type: Optional[PlaybookType] = Field(default=None, description="Linked object type") - name: Optional[str] = Field(default="", description="Executed playbook name") - user_input: Optional[str] = Field(default="", description="Initial or follow-up user input") - user: Optional[Union[List[AccountModel], AccountModel, str]] = Field(default=None, description="Playbook requester") + id: Optional[str] = Field(default=None, description="剧本记录 ID (Playbook record ID)") + source_rowid: Optional[str] = Field(default="", description="触发源行 ID (Trigger source row ID)") + source_id: Optional[str] = Field(default="", description="触发源记录 ID (Trigger source record ID e.g. case_00000_1,alert_000001,artifact_000001)") + type: Optional[PlaybookType] = Field(default=None, description="关联对象类型 (Linked object type)") + name: Optional[str] = Field(default="", description="执行剧本名称 (Executed playbook name)") + user_input: Optional[str] = Field(default="", description="初始或后续用户输入 (Initial or follow-up user input)") + user: Optional[Union[List[AccountModel], AccountModel, str]] = Field(default=None, description="剧本请求者 (Playbook requester)") - job_status: Optional[PlaybookJobStatus] = Field(default=None, description="Background job status") - job_id: Optional[str] = Field(default="", description="Background job ID") - remark: Optional[str] = Field(default="", description="Execution remark") + job_status: Optional[PlaybookJobStatus] = Field(default=None, description="后台任务状态 (Background job status)") + job_id: Optional[str] = Field(default="", description="后台任务 ID (Background job ID)") + remark: Optional[str] = Field(default="", description="执行备注 (Execution remark)") # 关联表 - messages: Optional[List[Union[MessageModel, str]]] = Field(default=None, description="Execution message history") + messages: Optional[List[Union[MessageModel, str]]] = Field(default=None, description="执行消息历史 (Execution message history)") class KnowledgeModel(BaseSystemModel): - title: Optional[str] = Field(default="", description="Knowledge title") - body: Optional[str] = Field(default="", description="Knowledge content") - using: Optional[bool] = Field(default=False, description="Currently in use") - action: Optional[KnowledgeAction] = Field(default=None, description="Knowledge action") - source: Optional[KnowledgeSource] = Field(default=None, description="Knowledge source") - tags: Optional[List[str]] = Field(default=[], description="Knowledge tags", json_schema_extra={"type": 2}) + title: Optional[str] = Field(default="", description="知识标题 (Knowledge title)") + body: Optional[str] = Field(default="", description="知识内容 (Knowledge content)") + using: Optional[bool] = Field(default=False, description="当前正在使用 (Currently in use)") + action: Optional[KnowledgeAction] = Field(default=None, description="知识操作 (Knowledge action)") + source: Optional[KnowledgeSource] = Field(default=None, description="知识来源 (Knowledge source)") + tags: Optional[List[str]] = Field(default=[], description="知识标签 (Knowledge tags)", json_schema_extra={"type": 2}) class EnrichmentModel(BaseSystemModel): ai_exclude_fields: ClassVar[set[str]] = {'ownerid', 'caid', 'uaid'} - id: Optional[str] = Field(default=None) - name: Optional[str] = Field(default="", description="Enrichment name") - type: Optional[str] = Field(default="Other", description="Enrichment type", json_schema_extra={"type": 2}) - provider: Optional[str] = Field(default="Other", description="Enrichment provider", json_schema_extra={"type": 2}) - value: Optional[str] = Field(default="", description="Enrichment value") - src_url: Optional[str] = Field(default="", description="Enrichment source URL") - desc: Optional[str] = Field(default="", description="Enrichment summary") - data: Optional[str] = Field(default="", description="Detailed enrichment JSON") + id: Optional[str] = Field(default=None, description="富化记录 ID (Enrichment record ID)") + name: Optional[str] = Field(default="", description="富化名称 (Enrichment name)") + type: Optional[str] = Field(default="Other", description="富化类型 (Enrichment type)", json_schema_extra={"type": 2}) + provider: Optional[str] = Field(default="Other", description="富化提供商 (Enrichment provider)", json_schema_extra={"type": 2}) + value: Optional[str] = Field(default="", description="富化值 (Enrichment value)") + src_url: Optional[str] = Field(default="", description="富化来源 URL (Enrichment source URL)") + desc: Optional[str] = Field(default="", description="富化摘要 (Enrichment summary)") + data: Optional[str] = Field(default="", description="详细富化 JSON (Detailed enrichment JSON)") class TicketModel(BaseSystemModel): ai_exclude_fields: ClassVar[set[str]] = {'ownerid', 'caid', 'uaid'} status: Optional[TicketStatus] = Field( - default=None, description="External ticket status") - type: Optional[TicketType] = Field(default=None, description="External ticket type", + default=None, description="外部工单状态 (External ticket status)") + type: Optional[TicketType] = Field(default=None, description="外部工单类型 (External ticket type)", json_schema_extra={"type": 2}) - title: Optional[str] = Field(default="", description="Ticket title") - uid: Optional[str] = Field(default="", description="External ticket ID") - src_url: Optional[str] = Field(default="", description="External ticket URL") + title: Optional[str] = Field(default="", description="工单标题 (Ticket title)") + uid: Optional[str] = Field(default="", description="外部工单 ID (External ticket ID)") + src_url: Optional[str] = Field(default="", description="外部工单 URL (External ticket URL)") # 反向关联 - case: Optional[List[Union[CaseModel, str]]] = Field(default=None, description="Linked case rowid") + case: Optional[List[Union[CaseModel, str]]] = Field(default=None, description="关联案例行 ID (Linked case rowid)") class ArtifactModel(BaseSystemModel): ai_exclude_fields: ClassVar[set[str]] = {'ownerid', 'caid', 'uaid'} - id: Optional[str] = Field(default=None) - name: Optional[str] = Field(default="", description="Artifact name") + id: Optional[str] = Field(default=None, description="痕迹记录 ID (Artifact record ID)") + name: Optional[str] = Field(default="", description="痕迹名称 (Artifact name)") type: Optional[ArtifactType] = Field( - default=None, description="Artifact type") + default=None, description="痕迹类型 (Artifact type)") role: Optional[ArtifactRole] = Field(default=None, - description="Artifact role in event") - owner: Optional[str] = Field(default="", description="Owning system or user") - value: Optional[str] = Field(default="", description="Artifact value") - reputation_provider: Optional[str] = Field(default="", description="Threat intel provider", json_schema_extra={"type": 2}) + description="痕迹在事件中的角色 (Artifact role in event)") + owner: Optional[str] = Field(default="", description="所属系统或用户 (Owning system or user)") + value: Optional[str] = Field(default="", description="痕迹值 (Artifact value)") + reputation_provider: Optional[str] = Field(default="", description="威胁情报提供商 (Threat intel provider)", json_schema_extra={"type": 2}) reputation_score: Optional[ArtifactReputationScore] = Field( - default=None, description="Artifact reputation") + default=None, description="痕迹信誉 (Artifact reputation)") # 反向关联 - alert: Optional[List[Union[AlertModel, str]]] = Field(default=None, description="Linked alert rowid") + alert: Optional[List[Union[AlertModel, str]]] = Field(default=None, description="关联告警行 ID (Linked alert rowid)") # 关联表 - enrichments: Optional[List[Union[EnrichmentModel, str]]] = Field(default=None, description="Artifact enrichments") # None 时表示无需处理,[] 时表示要将 link 清空 - - # playbooks: Optional[Any] = "" # 内部字段 - # playbook: Optional[Literal['TI Enrichment By AlienVaultOTX', 'TI Enrichment By Mock', None]] = None # 内部字段 - # user_input: Optional[str] = "" # 内部字段 + enrichments: Optional[List[Union[EnrichmentModel, str]]] = Field(default=None, description="痕迹富化 (Artifact enrichments)") # None 时表示无需处理,[] 时表示要将 link 清空 class AlertModel(BaseSystemModel): - ai_exclude_fields: ClassVar[set[str]] = {'ownerid', 'caid', 'uaid', "comment_ai", "attachments"} - id: Optional[str] = Field(default=None) + ai_exclude_fields: ClassVar[set[str]] = {'ownerid', 'caid', 'uaid', "comment_ai", "attachments", "raw_data", "unmapped"} + id: Optional[str] = Field(default=None, description="告警记录 ID (Alert record ID)") severity: Optional[Severity] = Field(default=None, - description="Source-defined severity") - title: Optional[str] = Field(default="", description="Alert title") - impact: Optional[ImpactLevel] = Field(default=None, description="Potential impact") + description="源定义严重程度 (Source-defined severity)") + title: Optional[str] = Field(default="", description="告警标题 (Alert title)") + impact: Optional[ImpactLevel] = Field(default=None, description="潜在影响 (Potential impact)") disposition: Optional[Disposition] = Field( - default=None, description="Source disposition") + default=None, description="源处置结果 (Source disposition)") action: Optional[AlertAction] = Field(default=None, - description="Observed action") + description="观测到的动作 (Observed action)") confidence: Optional[Confidence] = Field(default=None, - description="True-positive confidence") - uid: Optional[str] = Field(default="", description="Alert unique ID") - labels: Optional[List[str]] = Field(default=[], description="Alert labels", json_schema_extra={"type": 2}) - desc: Optional[str] = Field(default="", description="Alert description") + description="真阳性置信度 (True-positive confidence)") + uid: Optional[str] = Field(default="", description="告警唯一 ID (Alert unique ID)") + labels: Optional[List[str]] = Field(default=[], description="告警标签 (Alert labels)", json_schema_extra={"type": 2}) + desc: Optional[str] = Field(default="", description="告警描述 (Alert description)") - first_seen_time: Optional[Union[datetime, str]] = Field(default=None, description="First observed time") - last_seen_time: Optional[Union[datetime, str]] = Field(default=None, description="Last observed time") + first_seen_time: Optional[Union[datetime, str]] = Field(default=None, description="首次观测时间 (First observed time)") + last_seen_time: Optional[Union[datetime, str]] = Field(default=None, description="最后观测时间 (Last observed time)") - rule_id: Optional[str] = Field(default="", description="Trigger rule ID") - rule_name: Optional[str] = Field(default="", description="Trigger rule name") - correlation_uid: Optional[str] = Field(default="", description="Event correlation ID") - count: Optional[Union[int, str]] = Field(default=None, description="Aggregated event count") + rule_id: Optional[str] = Field(default="", description="触发规则 ID (Trigger rule ID)") + rule_name: Optional[str] = Field(default="", description="触发规则名称 (Trigger rule name)") + correlation_uid: Optional[str] = Field(default="", description="事件关联 ID (Event correlation ID)") + count: Optional[Union[int, str]] = Field(default=None, description="聚合事件计数 (Aggregated event count)") - src_url: Optional[str] = Field(default="", description="Source alert URL") - source_uid: Optional[str] = Field(default="", description="Source product ID") - data_sources: Optional[List[str]] = Field(default=[], description="Underlying data sources") + src_url: Optional[str] = Field(default="", description="原始告警 URL (Source alert URL)") + source_uid: Optional[str] = Field(default="", description="原始产品 ID (Source product ID)") + data_sources: Optional[List[str]] = Field(default=[], description="基础数据源 (Underlying data sources)") - analytic_name: Optional[str] = Field(default="", description="Analytic engine name") + analytic_name: Optional[str] = Field(default="", description="分析引擎名称 (Analytic engine name)") analytic_type: Optional[AlertAnalyticType] = Field( - default=None, description="Analytic engine type") - analytic_state: Optional[AlertAnalyticState] = Field(default=None, description="Analytic rule state") - analytic_desc: Optional[str] = Field(default="", description="Analytic rule description") + default=None, description="分析引擎类型 (Analytic engine type)") + analytic_state: Optional[AlertAnalyticState] = Field(default=None, description="分析规则状态 (Analytic rule state)") + analytic_desc: Optional[str] = Field(default="", description="分析规则描述 (Analytic rule description)") - tactic: Optional[str] = Field(default="", description="Mapped MITRE tactic") - technique: Optional[str] = Field(default="", description="Mapped MITRE technique") - sub_technique: Optional[str] = Field(default="", description="Mapped MITRE sub-technique") - mitigation: Optional[str] = Field(default="", description="Suggested mitigation") + tactic: Optional[str] = Field(default="", description="映射的 MITRE 战术 (Mapped MITRE tactic)") + technique: Optional[str] = Field(default="", description="映射的 MITRE 技术 (Mapped MITRE technique)") + sub_technique: Optional[str] = Field(default="", description="映射的 MITRE 子技术 (Mapped MITRE sub-technique)") + mitigation: Optional[str] = Field(default="", description="建议的缓解措施 (Suggested mitigation)") product_category: Optional[ProductCategory] = Field(default=None, - description="Source product category") - product_vendor: Optional[str] = Field(default=None, description="Source vendor", json_schema_extra={"type": 2}) - product_name: Optional[str] = Field(default=None, description="Source product name", json_schema_extra={"type": 2}) - product_feature: Optional[str] = Field(default=None, description="Source product feature", json_schema_extra={"type": 2}) + description="原始产品类别 (Source product category)") + product_vendor: Optional[str] = Field(default=None, description="原始厂商 (Source vendor)", json_schema_extra={"type": 2}) + product_name: Optional[str] = Field(default=None, description="原始产品名称 (Source product name)", json_schema_extra={"type": 2}) + product_feature: Optional[str] = Field(default=None, description="原始产品功能 (Source product feature)", json_schema_extra={"type": 2}) - policy_name: Optional[str] = Field(default="", description="Trigger policy name") + policy_name: Optional[str] = Field(default="", description="触发策略名称 (Trigger policy name)") policy_type: Optional[AlertPolicyType] = Field(default=None, - description="Trigger policy type") - policy_desc: Optional[str] = Field(default="", description="Trigger policy description") + description="触发策略类型 (Trigger policy type)") + policy_desc: Optional[str] = Field(default="", description="触发策略描述 (Trigger policy description)") - risk_level: Optional[AlertRiskLevel] = Field(default=None, description="Assessed risk level") - risk_details: Optional[str] = Field(default="", description="Risk assessment details") + risk_level: Optional[AlertRiskLevel] = Field(default=None, description="评估的风险等级 (Assessed risk level)") + risk_details: Optional[str] = Field(default="", description="风险评估详情 (Risk assessment details)") status: Optional[AlertStatus] = Field(default=None, - description="Alert handling status") - status_detail: Optional[str] = Field(default="", description="Handling status details") - remediation: Optional[str] = Field(default="", description="Remediation advice or record") + description="告警处理状态 (Alert handling status)") + status_detail: Optional[str] = Field(default="", description="处理状态详情 (Handling status details)") + remediation: Optional[str] = Field(default="", description="处置建议或记录 (Remediation advice or record)") - comment: Optional[str] = Field(default="", description="Analyst comment") + comment: Optional[str] = Field(default="", description="分析师注释 (Analyst comment)") - unmapped: Optional[str] = Field(default="", description="Raw unmapped fields") + unmapped: Optional[str] = Field(default="", description="原始未映射字段 (Raw unmapped fields)") - raw_data: Optional[str] = Field(default="", description="Raw alert log JSON") + raw_data: Optional[str] = Field(default="", description="原始告警日志 JSON (Raw alert log JSON)") - attachments: Optional[Union[List[AttachmentModel], str]] = Field(default=[], description="Alert attachments") + attachments: Optional[Union[List[AttachmentModel], str]] = Field(default=[], description="告警附件 (Alert attachments)") # AI字段 - severity_ai: Optional[Severity] = Field(default=None, description="AI-assessed severity") - confidence_ai: Optional[Confidence] = Field(default=None, description="AI-assessed confidence") - comment_ai: Optional[str] = Field(default="", description="AI-generated comment") + severity_ai: Optional[Severity] = Field(default=None, description="AI 评估严重程度 (AI-assessed severity)") + confidence_ai: Optional[Confidence] = Field(default=None, description="AI 评估置信度 (AI-assessed confidence)") + comment_ai: Optional[str] = Field(default="", description="AI 生成的注释 (AI-generated comment)") # 反向关联 - case: Optional[List[Union[CaseModel, str]]] = Field(default=None, description="Linked case rowid") + case: Optional[List[Union[CaseModel, str]]] = Field(default=None, description="关联案例行 ID (Linked case rowid)") # 关联表 - artifacts: Optional[List[Union[ArtifactModel, str]]] = Field(default=None, description="Extracted artifacts") - enrichments: Optional[List[Union[EnrichmentModel, str]]] = Field(default=None, description="Alert enrichments") + artifacts: Optional[List[Union[ArtifactModel, str]]] = Field(default=None, description="提取的痕迹 (Extracted artifacts)") + enrichments: Optional[List[Union[AlertModel, str]]] = Field(default=None, description="告警富化 (Alert enrichments)") @field_validator('attachments', mode='before') def handle_attachments(cls, v): @@ -712,59 +682,59 @@ class CaseModel(BaseSystemModel): ai_exclude_fields: ClassVar[set[str]] = {'ownerid', 'caid', 'uaid', "workbook", "summary_ai", "comment_ai", "attack_stage_ai", "severity_ai", "confidence_ai", "threat_hunting_report_ai"} - id: Optional[str] = Field(default=None) - title: Optional[str] = Field(default="", description="Case title") + id: Optional[str] = Field(default=None, description="案例记录 ID (Case record ID)") + title: Optional[str] = Field(default="", description="案例标题 (Case title)") severity: Optional[Severity] = Field(default=None, - description="Analyst-assessed severity") - impact: Optional[ImpactLevel] = Field(default=None, description="Analyst-assessed impact") - priority: Optional[CasePriority] = Field(default=None, description="Response priority") - src_url: Optional[str] = Field(default="", description="Source case URL") - confidence: Optional[Confidence] = Field(default=None, description="Analyst-assessed confidence") - description: Optional[str] = Field(default="", description="Case description") + description="分析师评估严重程度 (Analyst-assessed severity)") + impact: Optional[ImpactLevel] = Field(default=None, description="分析师评估影响 (Analyst-assessed impact)") + priority: Optional[CasePriority] = Field(default=None, description="响应优先级 (Response priority)") + src_url: Optional[str] = Field(default="", description="原始案例 URL (Source case URL)") + confidence: Optional[Confidence] = Field(default=None, description="分析师评估置信度 (Analyst-assessed confidence)") + description: Optional[str] = Field(default="", description="案例描述 (Case description)") category: Optional[ProductCategory] = Field(default=None, - description="Case category") - tags: Optional[List[str]] = Field(default=[], description="Case tags", json_schema_extra={"type": 2}) + description="案例类别 (Case category)") + tags: Optional[List[str]] = Field(default=[], description="案例标签 (Case tags)", json_schema_extra={"type": 2}) status: Optional[CaseStatus] = Field(default=None, - description="Case handling status") - assignee_l1: Optional[Union[List[AccountModel], AccountModel, str]] = Field(default=None, description="Assigned L1 analyst") - acknowledged_time: Optional[Union[datetime, str]] = Field(default=None, description="L1 first acknowledged time") - comment: Optional[str] = Field(default="", description="Case analyst comment") - attachments: Optional[List[Union[AttachmentModel, AttachmentCreateModel]]] = Field(default=[], description="Case attachments") + description="案例处理状态 (Case handling status)") + assignee_l1: Optional[Union[List[AccountModel], AccountModel, str]] = Field(default=None, description="分配的 L1 分析师 (Assigned L1 analyst)") + acknowledged_time: Optional[Union[datetime, str]] = Field(default=None, description="L1 首次接手时间 (L1 first acknowledged time)") + comment: Optional[str] = Field(default="", description="案例分析师注释 (Case analyst comment)") + attachments: Optional[List[Union[AttachmentModel, AttachmentCreateModel]]] = Field(default=[], description="案例附件 (Case attachments)") - assignee_l2: Optional[Union[List[AccountModel], AccountModel, str]] = Field(default=None, description="Assigned or escalated L2 analyst") - assignee_l3: Optional[Union[List[AccountModel], AccountModel, str]] = Field(default=None, description="Assigned or escalated L3 analyst") - closed_time: Optional[Union[datetime, str]] = Field(default=None, description="Case closed time") + assignee_l2: Optional[Union[List[AccountModel], AccountModel, str]] = Field(default=None, description="分配或升级的 L2 分析师 (Assigned or escalated L2 analyst)") + assignee_l3: Optional[Union[List[AccountModel], AccountModel, str]] = Field(default=None, description="分配或升级的 L3 分析师 (Assigned or escalated L3 analyst)") + closed_time: Optional[Union[datetime, str]] = Field(default=None, description="案例关闭时间 (Case closed time)") verdict: Optional[CaseVerdict] = Field( - default=None, description="Final verdict") - summary: Optional[str] = Field(default="", description="Closure summary") + default=None, description="最终判定结果 (Final verdict)") + summary: Optional[str] = Field(default="", description="结案摘要 (Closure summary)") - correlation_uid: Optional[str] = Field(default="", description="Case correlation ID") + correlation_uid: Optional[str] = Field(default="", description="案例关联 ID (Case correlation ID)") - workbook: Optional[str] = Field(default="", description="Investigation workbook") + workbook: Optional[str] = Field(default="", description="调查工作手册 (Investigation workbook)") # ai 字段 - attack_stage_ai: Optional[AttackStage] = Field(default="", description="AI-assessed attack stage") + attack_stage_ai: Optional[AttackStage] = Field(default="", description="AI 评估攻击阶段 (AI-assessed attack stage)") severity_ai: Optional[Severity] = Field(default=None, - description="AI-assessed severity") - confidence_ai: Optional[Confidence] = Field(default=None, description="AI-assessed confidence") - comment_ai: Optional[str] = Field(default="", description="AI-generated comment") - summary_ai: Optional[str] = Field(default="", description="AI-generated closure summary") - verdict_ai: Optional[CaseVerdict] = Field(default=None, description="AI-generated Final verdict") - threat_hunting_report_ai: Optional[str] = Field(default="", description="AI-generated hunting report") + description="AI 评估严重程度 (AI-assessed severity)") + confidence_ai: Optional[Confidence] = Field(default=None, description="AI 评估置信度 (AI-assessed confidence)") + comment_ai: Optional[str] = Field(default="", description="AI 生成的注释 (AI-generated comment)") + summary_ai: Optional[str] = Field(default="", description="AI 生成的结案摘要 (AI-generated closure summary)") + verdict_ai: Optional[CaseVerdict] = Field(default=None, description="AI 生成的最终判定结果 (AI-generated Final verdict)") + threat_hunting_report_ai: Optional[str] = Field(default="", description="AI 生成的威胁狩猎报告 (AI-generated hunting report)") # 公式计算字段 - start_time_calc: Optional[Any] = Field(default=None, description="Calculated start time") - end_time_calc: Optional[Any] = Field(default=None, description="Calculated end time") - detect_time_calc: Optional[Any] = Field(default=None, description="Calculated detect time") - acknowledge_time_calc: Optional[Any] = Field(default=None, description="Calculated acknowledge time") - respond_time_calc: Optional[Any] = Field(default=None, description="Calculated response time") + start_time_calc: Optional[Any] = Field(default=None, description="计算的开始时间 (Calculated start time)") + end_time_calc: Optional[Any] = Field(default=None, description="计算的结束时间 (Calculated end time)") + detect_time_calc: Optional[Any] = Field(default=None, description="计算的检测时间 (Calculated detect time)") + acknowledge_time_calc: Optional[Any] = Field(default=None, description="计算的接手时间 (Calculated acknowledge time)") + respond_time_calc: Optional[Any] = Field(default=None, description="计算的响应时间 (Calculated response time)") # 关联表 - tickets: Optional[List[Union[TicketModel, str]]] = Field(default=None, description="Linked external tickets") - enrichments: Optional[List[Union[EnrichmentModel, str]]] = Field(default=None, description="Case enrichments") - alerts: Optional[List[Union[AlertModel, str]]] = Field(default=None, description="Merged alerts") + tickets: Optional[List[Union[TicketModel, str]]] = Field(default=None, description="关联外部工单 (Linked external tickets)") + enrichments: Optional[List[Union[EnrichmentModel, str]]] = Field(default=None, description="案例富化 (Case enrichments)") + alerts: Optional[List[Union[AlertModel, str]]] = Field(default=None, description="合并的告警 (Merged alerts)") @field_validator('attachments', mode='before') def handle_attachments(cls, v):