From aaf024262274ef77fff3c32948ece70b1a02c9af Mon Sep 17 00:00:00 2001 From: rootkit Date: Thu, 12 Feb 2026 04:41:11 +0800 Subject: [PATCH] update --- ...ivilege-Escalation-via-AttachUserPolicy.py | 107 ++++---- PLUGINS/Mock/SIRP/rule.py | 239 ------------------ PLUGINS/SIRP/grouprule.py | 195 +++++++++----- 3 files changed, 190 insertions(+), 351 deletions(-) delete mode 100644 PLUGINS/Mock/SIRP/rule.py 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 8f0b8e9..42f47c6 100644 --- a/MODULES/Cloud-01-AWS-IAM-Privilege-Escalation-via-AttachUserPolicy.py +++ b/MODULES/Cloud-01-AWS-IAM-Privilege-Escalation-via-AttachUserPolicy.py @@ -12,6 +12,7 @@ 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, CorrelationStrategy from PLUGINS.SIRP.sirpapi import Alert from PLUGINS.SIRP.sirpmodel import AlertModel, ArtifactModel, ArtifactType, ArtifactRole, Severity, AlertStatus, AlertAnalyticType, ProductCategory, Confidence, \ ImpactLevel, AlertRiskLevel, Disposition, AlertAction, AlertPolicyType @@ -85,54 +86,6 @@ class Module(LanggraphModule): event_time_formatted = event_time if event_time else get_current_time_str() - correlation_uid = f"{account_id}-{principal_user}-{target_user}-{event_id}" - - 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" - ) - artifacts: List[ArtifactModel] = [] artifacts.append(ArtifactModel( @@ -185,6 +138,64 @@ class Module(LanggraphModule): name="Access Key ID" )) + correlation_config = CorrelationConfig( + rule_id=self.module_name, + strategy=CorrelationStrategy.BY_ACTOR_AND_TARGET, + time_window="24h", + case_title_template="AWS IAM Privilege Escalation: {actor} → {target}" + ) + group_rule = GroupRule(config=correlation_config) + correlation_uid = group_rule.generate_correlation_uid( + artifacts=artifacts, + 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 return {"alert": alert_model} diff --git a/PLUGINS/Mock/SIRP/rule.py b/PLUGINS/Mock/SIRP/rule.py deleted file mode 100644 index 37ebe37..0000000 --- a/PLUGINS/Mock/SIRP/rule.py +++ /dev/null @@ -1,239 +0,0 @@ -from PLUGINS.SIRP.grouprule import GroupRule - -rule_list = [ - GroupRule( - rule_id="EDR-Rule-11-Suspicious-Process-Spawned-by-Office", - rule_name="Suspicious process spawned by Office application", - deduplication_fields=["hostname", "process_name"], - source="EDR" - ), - GroupRule( - rule_id="EDR-Rule-21-CobaltStrike-Beacon-Detected", - rule_name="Cobalt Strike C2 beacon detected", - deduplication_fields=["hostname"], - deduplication_window="1h", - source="EDR" - ), - GroupRule( - rule_id="EDR-Rule-07-Credential-Dumping-LSASS", - rule_name="LSASS memory credential dump", - deduplication_fields=["hostname", "target_process"], - source="EDR" - ), - GroupRule( - rule_id="EDR-Rule-01-Suspicious-PowerShell-Execution", - rule_name="Suspicious PowerShell command execution", - deduplication_fields=["hostname"], - deduplication_window="1h", - source="EDR" - ), - GroupRule( - rule_id="EDR-Rule-02-Unusual-Network-Connection-to-External", - rule_name="Unusual outbound network connection", - deduplication_fields=["hostname", "destination_ip"], - source="EDR" - ), - GroupRule( - rule_id="NDR-Rule-05-Suspect-C2-Communication", - rule_name="Suspicious command and control (C2) communication", - deduplication_fields=["hostname"], - source="NDR" - ), - GroupRule( - rule_id="NDR-Rule-12-Lateral-Movement-Attempt", - rule_name="Host-to-host lateral movement attempt", - deduplication_fields=["hostname", "destination_ip"], - deduplication_window="1h", - source="NDR" - ), - GroupRule( - rule_id="NDR-Rule-15-Unauthorized-Data-Exfiltration", - rule_name="Anomalous data exfiltration", - deduplication_fields=["hostname", "data_volume"], - source="NDR" - ), - GroupRule( - rule_id="NDR-Rule-01-C2-Beaconing", - rule_name="C2 beaconing traffic", - deduplication_fields=["hostname", "destination_ip"], - deduplication_window="1h", - source="NDR" - ), - GroupRule( - rule_id="NDR-Rule-02-Internal-Port-Scan", - rule_name="Internal port scan", - deduplication_fields=["source_ip", "scan_type"], - source="NDR" - ), - GroupRule( - rule_id="DLP-Rule-08-Financial-Record-Transfer-to-USB", - rule_name="Financial records transferred to removable device", - deduplication_fields=["hostname", "username"], - source="DLP" - ), - GroupRule( - rule_id="DLP-Rule-09-Source-Code-Upload-to-Pastebin", - rule_name="Source code uploaded to public site", - deduplication_fields=["hostname", "username"], - deduplication_window="1h", - source="DLP" - ), - GroupRule( - rule_id="DLP-Rule-10-Health-Information-Transfer", - rule_name="Protected health information (PHI) transfer", - deduplication_fields=["hostname", "username", "data_classification"], - source="DLP" - ), - GroupRule( - rule_id="DLP-Rule-11-Leaked-API-Key-in-Code", - rule_name="API key leaked", - deduplication_fields=["hostname", "username", "repository"], - source="DLP" - ), - GroupRule( - rule_id="DLP-Rule-12-Internal-SSN-Transfer", - rule_name="", - deduplication_fields=["hostname", "username"], - deduplication_window="1h", - source="DLP" - ), - GroupRule( - rule_id="ES-Rule-01-Phishing-URL-Detected", - rule_name="Phishing URL detected in email", - deduplication_fields=["sender_email"], - source="Email" - ), - GroupRule( - rule_id="ES-Rule-02-Malicious-Attachment-Detected", - rule_name="Malicious attachment detected in email", - deduplication_fields=["sender_email"], - source="Email" - ), - GroupRule( - rule_id="ES-Rule-03-BEC-Spoofing-CEO", - rule_name="Business Email Compromise (BEC) - CEO impersonation", - deduplication_fields=["sender_email", "subject"], - source="Email" - ), - GroupRule( - rule_id="ES-Rule-04-Credential-Phishing-Page", - rule_name="Credential phishing page link", - deduplication_fields=["sender_email", "subject"], - deduplication_window="10m", - source="Email" - ), - GroupRule( - rule_id="ES-Rule-05-Fileless-Malware-Detected", - rule_name="Fileless malware detected in email", - deduplication_fields=["sender_email"], - source="Email" - ), - GroupRule( - rule_id="OT-Rule-01-PLC-Configuration-Change", - rule_name="Unauthorized PLC configuration change", - deduplication_fields=["device_id"], - source="OT" - ), - GroupRule( - rule_id="OT-Rule-02-Unusual-Protocol-Activity", - rule_name="Suspicious protocol activity in SCADA network", - deduplication_fields=["source_device"], - source="OT" - ), - GroupRule( - rule_id="OT-Rule-03-Controller-Stop-Command", - rule_name="Controller received stop command", - deduplication_fields=["device_id"], - source="OT" - ), - GroupRule( - rule_id="PROXY-Rule-01-Malware-Download-Blocked", - rule_name="Malicious download blocked", - deduplication_fields=["source_ip"], - source="Proxy" - ), - GroupRule( - rule_id="PROXY-Rule-02-C2-Communication-Blocked", - rule_name="C2 communication blocked", - deduplication_fields=["source_ip"], - source="Proxy" - ), - GroupRule( - rule_id="PROXY-Rule-03-Phishing-URL-Detected", - rule_name="Access to phishing site blocked", - deduplication_fields=["source_ip"], - source="Proxy" - ), - GroupRule( - rule_id="UEBA-Rule-01-Lateral-Movement-Spike", - rule_name="Anomalous nighttime lateral movement", - deduplication_fields=["username"], - source="UEBA" - ), - GroupRule( - rule_id="UEBA-Rule-02-Unusual-Data-Volume-Download", - rule_name="Abnormal data exfiltration volume", - deduplication_fields=["username", "data_destination"], - source="UEBA" - ), - GroupRule( - rule_id="UEBA-Rule-03-Account-Brute-Force-Multiple-Sources", - rule_name="Multi-source account brute force", - deduplication_fields=["target_username"], - source="UEBA" - ), - GroupRule( - rule_id="TI-Rule-01-Malicious-IP-Inbound", - rule_name="Inbound connection from malicious IP", - deduplication_fields=["destination_ip"], - source="TI" - ), - GroupRule( - rule_id="TI-Rule-02-C2-Domain-Outbound", - rule_name="Internal host attempted connection to C2 domain", - deduplication_fields=["destination_domain"], - source="TI" - ), - GroupRule( - rule_id="TI-Rule-03-Malicious-File-Hash-Match", - rule_name="Internal file hash matched threat intelligence", - deduplication_fields=["hostname"], - source="TI" - ), - GroupRule( - rule_id="IAM-Rule-01-Excessive-Permission-Grant", - rule_name="Account permissions escalated abnormally", - deduplication_fields=["username", "platform"], - source="IAM" - ), - GroupRule( - rule_id="IAM-Rule-02-Impossible-Travel-Login", - rule_name="Impossible travel login", - deduplication_fields=["username"], - source="IAM" - ), - GroupRule( - rule_id="IAM-Rule-03-Brute-Force-Attack-Password-Spraying", - rule_name="Password spraying across multiple accounts", - deduplication_fields=["source_ip"], - source="IAM" - ), - GroupRule( - rule_id="CLOUD-AWS-IAM-01-Root-User-Activity", - rule_name="Root account activity", - deduplication_fields=["platform", "service"], - source="Cloud" - ), - GroupRule( - rule_id="CLOUD-AZ-COMPUTE-02-VM-Reconnaissance", - rule_name="Virtual machine internal reconnaissance", - deduplication_fields=["vm_name", "vm_ip"], - source="Cloud" - ), - GroupRule( - rule_id="CLOUD-GCP-STORAGE-03-Public-Bucket-Access", - rule_name="Public bucket access", - deduplication_fields=["bucket_name"], - source="Cloud" - ), -] diff --git a/PLUGINS/SIRP/grouprule.py b/PLUGINS/SIRP/grouprule.py index 44f90ec..2e07133 100644 --- a/PLUGINS/SIRP/grouprule.py +++ b/PLUGINS/SIRP/grouprule.py @@ -1,40 +1,66 @@ +import hashlib from datetime import datetime, timezone +from enum import StrEnum from typing import List, Dict, Any, Optional, Union +from pydantic import BaseModel, Field + + +class CorrelationStrategy(StrEnum): + BY_ACTOR = "by_actor" + BY_TARGET = "by_target" + BY_ACTOR_AND_TARGET = "by_actor_and_target" + BY_RULE = "by_rule" + BY_CUSTOM_FIELDS = "by_custom_fields" + + +class CorrelationConfig(BaseModel): + rule_id: str = Field(description="规则ID，用于区分不同类型的告警") + strategy: CorrelationStrategy = Field(default=CorrelationStrategy.BY_ACTOR_AND_TARGET, description="关联策略") + artifact_types: List[str] = Field(default=[], description="用于关联的artifact类型列表，为空时根据strategy自动选择") + artifact_roles: List[str] = Field(default=[], description="用于关联的artifact角色列表，为空时根据strategy自动选择") + time_window: str = Field(default="24h", description="时间窗口") + case_title_template: str = Field(default="", description="Case标题模板，支持{rule_name}, {actor}, {target}等变量") + include_rule_in_key: bool = Field(default=True, description="是否将规则ID包含在关联key中") + class GroupRule(object): - """ - 告警聚合规则,用于定义如何将多个告警(alert)聚合到同一个案件(case)中. - 基本方法: 根据规则ID、告警中的凭据(artifact)以及时间窗口,生成去重指纹(deduplication key).还有相同指纹的告警会被聚合到同一个案件中. - 还可以定义案件标题的模板,以便生成更具描述性的案件标题. - 方法在心智成本和聚合效果之间取得平衡,适用于绝大多数常见地告警聚合场景. - """ + VALID_WINDOWS = ['10m', '30m', '1h', '8h', '12h', '24h'] + + ROLE_PRIORITY = { + 'Actor': 1, + 'Target': 2, + 'Affected': 3, + 'Related': 4, + 'Unknown': 5, + 'Other': 6 + } def __init__(self, - rule_id: str, - rule_name: str, - deduplication_fields: List[str], - case_title_template: str = None, - deduplication_window: str = "24h", - source: str = "Default", - workbook: str = None, - follow_alert_severity: bool = True, - append_alert_tags: bool = True, - ): + config: Optional[CorrelationConfig] = None, + rule_id: str = None, + correlation_fields: List[str] = None, + correlation_window: str = "24h"): - self.rule_id = rule_id - self.rule_name = rule_name - self.deduplication_fields = deduplication_fields - self.case_title_template = case_title_template - self.source = source - self.workbook = workbook - self.follow_alert_severity = follow_alert_severity - self.append_alert_tags = append_alert_tags - - valid_windows = ['10m', '30m', '1h', '8h', '12h', '24h'] - if deduplication_window not in valid_windows: - raise ValueError(f"'{deduplication_window}' 不是一个有效的时间窗口选项.请从 {valid_windows} 中选择.") - self.deduplication_window = deduplication_window + if config: + self.rule_id = config.rule_id + self.strategy = config.strategy + self.artifact_types = config.artifact_types + self.artifact_roles = config.artifact_roles + self.time_window = config.time_window + self.case_title_template = config.case_title_template + self.include_rule_in_key = config.include_rule_in_key + else: + self.rule_id = rule_id or "" + self.strategy = CorrelationStrategy.BY_CUSTOM_FIELDS + self.artifact_types = correlation_fields or [] + self.artifact_roles = [] + self.time_window = correlation_window + self.case_title_template = "" + self.include_rule_in_key = True + + if self.time_window not in self.VALID_WINDOWS: + raise ValueError(f"'{self.time_window}' 不是一个有效的时间窗口选项. 请从 {self.VALID_WINDOWS} 中选择.") @staticmethod def _get_time_bucket(dt_object: datetime, window: str) -> datetime: @@ -51,42 +77,83 @@ class GroupRule(object): return dt_object.replace(hour=new_hour, minute=0, second=0, microsecond=0) return dt_object - def generate_deduplication_key(self, - artifacts: List[Dict[str, Any]], - timestamp: Optional[Union[int, float]] = None) -> str: - """ - 生成包含“时间桶”的去重指纹. + def _extract_artifacts_by_strategy(self, artifacts: List[Any]) -> List[Dict[str, str]]: + result = [] - :param artifacts: 事件中的凭据列表 - :param timestamp: (可选) 事件的UTC Unix时间戳 (整数或浮点数).如果为None,则使用当前系统时间. - :return: 包含时间桶的去重指纹 - """ - if timestamp is not None: - processing_dt = datetime.fromtimestamp(timestamp, tz=timezone.utc) - else: + role_map = { + CorrelationStrategy.BY_ACTOR: ['Actor'], + CorrelationStrategy.BY_TARGET: ['Target'], + CorrelationStrategy.BY_ACTOR_AND_TARGET: ['Actor', 'Target'], + CorrelationStrategy.BY_RULE: [], + CorrelationStrategy.BY_CUSTOM_FIELDS: self.artifact_roles if self.artifact_roles else [] + } + target_roles = role_map.get(self.strategy, []) + + for artifact in artifacts: + if hasattr(artifact, 'type') and hasattr(artifact, 'value') and hasattr(artifact, 'role'): + art_type = str(artifact.type.value) if hasattr(artifact.type, 'value') else str(artifact.type) + art_value = str(artifact.value) if artifact.value else "" + art_role = str(artifact.role.value) if hasattr(artifact.role, 'value') else str(artifact.role) + elif isinstance(artifact, dict): + art_type = str(artifact.get('type', '')) + art_value = str(artifact.get('value', '')) + art_role = str(artifact.get('role', 'Related')) + else: + continue + + if self.artifact_types and art_type not in self.artifact_types: + continue + + if target_roles and art_role not in target_roles: + continue + + result.append({ + 'type': art_type, + 'value': art_value, + 'role': art_role + }) + + result.sort(key=lambda x: (self.ROLE_PRIORITY.get(x['role'], 99), x['type'], x['value'])) + + return result + + def generate_correlation_uid(self, + artifacts: List[Any], + timestamp: Optional[Union[int, float, str, datetime]] = None, + rule_id_override: str = None) -> str: + + if timestamp is None: processing_dt = datetime.now(timezone.utc) - - time_bucket_dt = self._get_time_bucket(processing_dt, self.deduplication_window) - time_bucket_str = time_bucket_dt.strftime('%Y-%m-%dT%H:%M:%S') - - key_parts = [self.rule_id, time_bucket_str] - artifacts_map = {art['type']: art['value'] for art in artifacts} - for field in sorted(self.deduplication_fields): - key_parts.append(artifacts_map.get(field, 'N/A')) - - return "_".join(key_parts) - - def generate_case_title(self, artifacts: List[Dict[str, Any]] = None) -> str: - if self.case_title_template is None: - title = self.rule_name - for art in artifacts: - if art.get('type') in self.deduplication_fields: - title = f"{title} {art['type']}:{art['value']}" - return title + elif isinstance(timestamp, datetime): + processing_dt = timestamp if timestamp.tzinfo else timestamp.replace(tzinfo=timezone.utc) + elif isinstance(timestamp, str): + try: + processing_dt = datetime.fromisoformat(timestamp.replace('Z', '+00:00')) + except ValueError: + processing_dt = datetime.now(timezone.utc) else: - template_values = {"rule_name": self.rule_name} - for art in artifacts: - template_values[art['type']] = art['value'] - title = self.case_title_template.format_map(template_values) - return title + processing_dt = datetime.fromtimestamp(timestamp, tz=timezone.utc) + + time_bucket_dt = self._get_time_bucket(processing_dt, self.time_window) + time_bucket_str = time_bucket_dt.strftime('%Y%m%d%H%M') + + key_parts = [] + + effective_rule_id = rule_id_override or self.rule_id + if self.include_rule_in_key and effective_rule_id: + key_parts.append(effective_rule_id) + + key_parts.append(time_bucket_str) + + extracted_artifacts = self._extract_artifacts_by_strategy(artifacts) + + for art in extracted_artifacts: + key_parts.append(f"{art['role']}:{art['type']}:{art['value']}") + + raw_key = "|".join(key_parts) + + hash_obj = hashlib.sha256(raw_key.encode('utf-8')) + short_hash = hash_obj.hexdigest()[:16] + + return f"corr-{short_hash}"