Add function to clean alert titles by removing BOM characters and normalizing encoding (#468)
taylor_socfortress committed
Jul 3, 2025 at 11:19 UTC
4a8f04149f6835eb73ff48ddae09ac6aa42958e0
1 file changed
+33
-1
backend/app/incidents/services/incident_alert.py
+33
-1
@@ -98,6 +98,34 @@ async def is_customer_code_valid(customer_code: str, session: AsyncSession) -> A
98
)
99
100
101
+def clean_alert_title(title: str) -> str:
102
+ """
103
+ Clean the alert title by removing BOM characters and normalizing encoding.
104
+
105
+ Args:
106
+ title (str): The raw alert title
107
+
108
+ Returns:
109
+ str: The cleaned alert title
110
+ """
111
+ if not title:
112
+ return title
113
+
114
+ # Remove BOM characters
115
+ title = title.replace("\ufeff", "") # UTF-8 BOM
116
+ title = title.replace("\ufffe", "") # UTF-16 BE BOM
117
+ title = title.replace("\xff\xfe", "") # UTF-16 LE BOM
118
+
119
+ # Strip whitespace and normalize
120
+ title = title.strip()
121
+
122
+ # Ensure proper UTF-8 encoding
123
+ if isinstance(title, str):
124
+ title = title.encode("utf-8", "ignore").decode("utf-8")
125
+
126
+ return title
127
+
128
+
129
async def get_single_alert_details(
130
alert_details: CreateAlertRequest,
131
) -> GenericAlertModel:
@@ -516,11 +544,15 @@ async def build_alert_payload(
544
detail=f"Field name {field_name} not found in alert payload",
545
)
546
547
+ # Clean the alert title to remove BOM and normalize encoding
548
+ raw_alert_title = alert_payload[field_names.alert_title_name] if field_names.alert_title_name in alert_payload else None
549
+ cleaned_alert_title = clean_alert_title(raw_alert_title) if raw_alert_title else None
550
+
551
return CreatedAlertPayload(
552
alert_context_payload=await build_alert_context_payload(alert_payload, field_names),
553
asset_payload=alert_payload[field_names.asset_name] if field_names.asset_name in alert_payload else None,
554
timefield_payload=alert_payload[field_names.timefield_name] if field_names.timefield_name in alert_payload else None,
523
- alert_title_payload=alert_payload[field_names.alert_title_name] if field_names.alert_title_name in alert_payload else None,
555
+ alert_title_payload=cleaned_alert_title,
556
ioc_payload=await build_ioc_payload(alert_payload, field_names),
557
source=syslog_type,
558
index_name=index_name,