update

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}
 

PLUGINS/Mock/SIRP/rule.py

@@ -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"
-    ),
-]

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}"