PhishingDedupPreprocessingRule Deprecated
Deprecated. Use the FindDuplicateEmailIncidents script instead.
python · Phishing
Source
import demistomock as demisto # noqa: F401 from CommonServerPython import * # noqa: F401 import dateutil # type: ignore from CommonServerUserPython import * import pandas as pd from bs4 import BeautifulSoup from sklearn.feature_extraction.text import CountVectorizer from numpy import dot from numpy.linalg import norm from email.utils import parseaddr import tldextract from urllib.parse import urlparse import re no_fetch_extract = tldextract.TLDExtract(suffix_list_urls=None) pd.options.mode.chained_assignment = None # default='warn' SIMILARITY_THRESHOLD = float(demisto.args().get("threshold", 0.97)) CLOSE_TO_SIMILAR_DISTANCE = 0.2 EMAIL_BODY_FIELD = "emailbody" EMAIL_SUBJECT_FIELD = "emailsubject" EMAIL_HTML_FIELD = "emailbodyhtml" FROM_FIELD = "emailfrom" FROM_DOMAIN_FIELD = "fromdomain" MERGED_TEXT_FIELD = "mereged_text" MIN_TEXT_LENGTH = 50 DEFAULT_ARGS = { "limit": "1000", "incidentTypes": "Phishing", "exsitingIncidentsLookback": "100 days ago", } FROM_POLICY_TEXT_ONLY = "TextOnly" FROM_POLICY_EXACT = "Exact" FROM_POLICY_DOMAIN = "Domain" FROM_POLICY = FROM_POLICY_TEXT_ONLY URL_REGEX = ( r"(?:(?:https?|ftp|hxxps?):\/\/|www\[?\.\]?|ftp\[?\.\]?)(?:[-\w\d]+\[?\.\]?)+[-\w\d]+(?::\d+)?" r"(?:(?:\/|\?)[-\w\d+&@#\/%=~_$?!\-:,.\(\);]*[\w\d+&@#\/%=~_$\(\);])?" ) IGNORE_INCIDENT_TYPE_VALUE = "None" def get_existing_incidents(input_args, current_incident_type): global DEFAULT_ARGS get_incidents_args = {} get_incidents_args["limit"] = input_args.get("limit", DEFAULT_ARGS["limit"]) if "exsitingIncidentsLookback" in input_args: get_incidents_args["fromDate"] = input_args["exsitingIncidentsLookback"] elif "exsitingIncidentsLookback" in DEFAULT_ARGS: get_incidents_args["fromDate"] = DEFAULT_ARGS["exsitingIncidentsLookback"] status_scope = input_args.get("statusScope", "All") query_components = [] if "query" in input_args: query_components.append(input_args["query"]) if status_scope == "ClosedOnly": query_components.append("status:closed") elif status_scope == "NonClosedOnly": query_components.append("-status:closed") elif status_scope == "All": pass else: return_error(f"Unsupported statusScope: {status_scope}") type_values = input_args.get("incidentTypes", current_incident_type) if type_values != IGNORE_INCIDENT_TYPE_VALUE: type_field = input_args.get("incidentTypeFieldName", "type") type_query = generate_incident_type_query_component(type_field, type_values) query_components.append(type_query) if len(query_components) > 0: get_incidents_args["query"] = " and ".join(f"({c})" for c in query_components) incidents_query_res = demisto.executeCommand("GetIncidentsByQuery", get_incidents_args) if is_error(incidents_query_res): return_error(get_error(incidents_query_res)) incidents = json.loads(incidents_query_res[-1]["Contents"]) return incidents def generate_incident_type_query_component(type_field_arg, type_values_arg): type_field = type_field_arg.strip() type_values = [x.strip() for x in type_values_arg.split(",")] types_unions = " ".join(f'"{t}"' for t in type_values) return f"{type_field}:({types_unions})" def extract_domain(address): global no_fetch_extract if address == "": return "" email_address = parseaddr(address)[1] ext = no_fetch_extract(email_address) return ext.domain def get_text_from_html(html): soup = BeautifulSoup(html, features="html.parser") # kill all script and style elements for script in soup(["script", "style"]): script.extract() # rip it out # get text text = soup.get_text() # break into lines and remove leading and trailing space on each lines = (line.strip() for line in text.splitlines()) # break multi-headlines into a line each chunks = (phrase.strip() for line in lines for phrase in line.split(" ")) # drop blank lines text = "\n".join(chunk for chunk in chunks if chunk) return text def eliminate_urls_extensions(text): urls_list = re.findall(URL_REGEX, text) for url in urls_list: parsed_uri = urlparse(url) url_with_no_path = f"{parsed_uri.scheme}://{parsed_uri.netloc}/" text = text.replace(url, url_with_no_path) return text def preprocess_text_fields(incident): email_body = email_subject = email_html = "" if EMAIL_BODY_FIELD in incident: email_body = incident[EMAIL_BODY_FIELD] if EMAIL_HTML_FIELD in incident: email_html = incident[EMAIL_HTML_FIELD] if EMAIL_SUBJECT_FIELD in incident: email_subject = incident[EMAIL_SUBJECT_FIELD] if isinstance(email_html, float): email_html = "" if email_body is None or isinstance(email_body, float) or email_body.strip() == "": email_body = get_text_from_html(email_html) if isinstance(email_subject, float): email_subject = "" text = eliminate_urls_extensions(email_subject + " " + email_body) return text def preprocess_incidents_df(existing_incidents): global MERGED_TEXT_FIELD, FROM_FIELD, FROM_DOMAIN_FIELD incidents_df = pd.DataFrame(existing_incidents) incidents_df["CustomFields"] = incidents_df["CustomFields"].fillna(value={}) custom_fields_df = incidents_df["CustomFields"].apply(pd.Series) unique_keys = [k for k in custom_fields_df if k not in incidents_df] custom_fields_df = custom_fields_df[unique_keys] incidents_df = pd.concat([incidents_df.drop("CustomFields", axis=1), custom_fields_df], axis=1).reset_index() incidents_df[MERGED_TEXT_FIELD] = incidents_df.apply(lambda x: preprocess_text_fields(x), axis=1) incidents_df = incidents_df[incidents_df[MERGED_TEXT_FIELD].str.len() >= MIN_TEXT_LENGTH] incidents_df.reset_index(inplace=True) # noqa: PD002 if FROM_FIELD in incidents_df: incidents_df[FROM_FIELD] = incidents_df[FROM_FIELD].fillna(value="") else: incidents_df[FROM_FIELD] = "" incidents_df[FROM_FIELD] = incidents_df[FROM_FIELD].apply(lambda x: x.strip()) incidents_df[FROM_DOMAIN_FIELD] = incidents_df[FROM_FIELD].apply(lambda address: extract_domain(address)) incidents_df["created"] = incidents_df["created"].apply(lambda x: dateutil.parser.parse(x)) # type: ignore return incidents_df def incident_has_text_fields(incident): text_fields = [EMAIL_SUBJECT_FIELD, EMAIL_HTML_FIELD, EMAIL_BODY_FIELD] custom_fields = incident.get("CustomFields", []) or [] if any(field in incident for field in text_fields): return True elif "CustomFields" in incident and any(field in custom_fields for field in text_fields): return True return False def filter_out_same_incident(existing_incidents_df, new_incident): same_id_mask = existing_incidents_df["id"] == new_incident["id"] existing_incidents_df = existing_incidents_df[~same_id_mask] return existing_incidents_df def filter_newer_incidents(existing_incidents_df, new_incident): new_incident_datetime = dateutil.parser.parse(new_incident["created"]) # type: ignore earlier_incidents_mask = existing_incidents_df["created"] < new_incident_datetime return existing_incidents_df[earlier_incidents_mask] def vectorize(text, vectorizer): return vectorizer.transform([text]).toarray()[0] def cosine_sim(a, b): return dot(a, b) / (norm(a) * norm(b)) def find_duplicate_incidents(new_incident, existing_incidents_df): global MERGED_TEXT_FIELD, FROM_POLICY new_incident_text = new_incident[MERGED_TEXT_FIELD] text = [new_incident_text] + existing_incidents_df[MERGED_TEXT_FIELD].tolist() vectorizer = CountVectorizer(token_pattern=r"(?u)\b\w\w+\b|!|\?|\"|\'").fit(text) new_incident_vector = vectorize(new_incident_text, vectorizer) existing_incidents_df["vector"] = existing_incidents_df[MERGED_TEXT_FIELD].apply(lambda x: vectorize(x, vectorizer)) existing_incidents_df["similarity"] = existing_incidents_df["vector"].apply(lambda x: cosine_sim(x, new_incident_vector)) if FROM_POLICY == FROM_POLICY_DOMAIN: mask = (existing_incidents_df[FROM_DOMAIN_FIELD] != "") & ( existing_incidents_df[FROM_DOMAIN_FIELD] == new_incident[FROM_DOMAIN_FIELD] ) existing_incidents_df = existing_incidents_df[mask] elif FROM_POLICY == FROM_POLICY_EXACT: mask = (existing_incidents_df[FROM_FIELD] != "") & (existing_incidents_df[FROM_FIELD] == new_incident[FROM_FIELD]) existing_incidents_df = existing_incidents_df[mask] existing_incidents_df["distance"] = existing_incidents_df["similarity"].apply(lambda x: 1 - x) tie_breaker_col = "id" try: existing_incidents_df["int_id"] = existing_incidents_df["id"].astype(int) tie_breaker_col = "int_id" except Exception: pass existing_incidents_df.sort_values(by=["distance", "created", tie_breaker_col], inplace=True) # noqa: PD002 if len(existing_incidents_df) > 0: return existing_incidents_df.iloc[0], existing_incidents_df.iloc[0]["similarity"] else: return None, None def return_entry(message, existing_incident=None, similarity=0): if existing_incident is None: similar_incident = {} else: similar_incident = { "rawId": existing_incident["id"], "id": existing_incident["id"], "name": existing_incident.get("name"), "similarity": similarity, } outputs = {"similarIncident": similar_incident, "isSimilarIncidentFound": existing_incident is not None} return_outputs(message, outputs) def close_new_incident_and_link_to_existing(new_incident, existing_incident, similarity): formatted_incident = format_similar_incident(existing_incident, similarity) message = tableToMarkdown(f"Duplicate incident found with similarity {similarity * 100:.1f}%", formatted_incident) if demisto.args().get("closeAsDuplicate", "true") == "true": res = demisto.executeCommand("CloseInvestigationAsDuplicate", {"duplicateId": existing_incident["id"]}) if is_error(res): return_error(res) message += "This incident (#{}) will be closed and linked to #{}.".format(new_incident["id"], existing_incident["id"]) return_entry(message, existing_incident.to_dict(), similarity) def create_new_incident(): return_entry("This incident is not a duplicate of an existing incident.") def format_similar_incident(incident, similairy): return { "Id": f"[{incident['id']}](#/Details/{incident['id']})", "Name": incident["name"], "Closed": incident.get("closed") != "0001-01-01T00:00:00Z", "Time": str(incident["created"]), "Email from": incident.get(demisto.args().get("emailFrom")), "Text Similarity": f"{similairy * 100:.1f}%", } def create_new_incident_low_similarity(existing_incident, similarity): message = "## This incident is not a duplicate of an existing incident.\n" if similarity > SIMILARITY_THRESHOLD - CLOSE_TO_SIMILAR_DISTANCE: formatted_incident = format_similar_incident(existing_incident, similarity) message += tableToMarkdown("Most similar incident found", formatted_incident) message += ( "The threshold for considering 2 incidents as duplicate is a similarity " f"of {SIMILARITY_THRESHOLD * 100:.1f}%.\n" ) message += ( "Therefore these 2 incidents will not be considered as duplicate and the current incident " "will remain active.\n" ) return_entry(message) def create_new_incident_no_text_fields(): text_fields = [EMAIL_BODY_FIELD, EMAIL_HTML_FIELD, EMAIL_SUBJECT_FIELD] message = "No text fields were found within this incident: {}.\n".format(",".join(text_fields)) message += "Incident will remain active." return_entry(message) def create_new_incident_too_short(): return_entry("Incident text after preprocessing is too short for deduplication. Incident will remain active.") def main(): global EMAIL_BODY_FIELD, EMAIL_SUBJECT_FIELD, EMAIL_HTML_FIELD, FROM_FIELD, MIN_TEXT_LENGTH, FROM_POLICY input_args = demisto.args() EMAIL_BODY_FIELD = input_args.get("emailBody", EMAIL_BODY_FIELD) EMAIL_SUBJECT_FIELD = input_args.get("emailSubject", EMAIL_SUBJECT_FIELD) EMAIL_HTML_FIELD = input_args.get("emailBodyHTML", EMAIL_HTML_FIELD) FROM_FIELD = input_args.get("emailFrom", FROM_FIELD) FROM_POLICY = input_args.get("fromPolicy", FROM_POLICY) new_incident = demisto.incidents()[0] existing_incidents = get_existing_incidents(input_args, new_incident.get("type", IGNORE_INCIDENT_TYPE_VALUE)) demisto.debug(f"found {len(existing_incidents)} incidents by query") if len(existing_incidents) == 0: create_new_incident() return None if not incident_has_text_fields(new_incident): create_new_incident_no_text_fields() return None new_incident_df = preprocess_incidents_df([new_incident]) if len(new_incident_df) == 0: # len(new_incident_df)==0 means new incident is too short create_new_incident_too_short() return None existing_incidents_df = preprocess_incidents_df(existing_incidents) existing_incidents_df = filter_out_same_incident(existing_incidents_df, new_incident) existing_incidents_df = filter_newer_incidents(existing_incidents_df, new_incident) if len(existing_incidents_df) == 0: create_new_incident() return None new_incident_preprocessed = new_incident_df.iloc[0].to_dict() duplicate_incident_row, similarity = find_duplicate_incidents(new_incident_preprocessed, existing_incidents_df) if duplicate_incident_row is None: create_new_incident() return None if similarity < SIMILARITY_THRESHOLD: create_new_incident_low_similarity(duplicate_incident_row, similarity) # noqa: RET503 else: return close_new_incident_and_link_to_existing(new_incident_df.iloc[0], duplicate_incident_row, similarity) if __name__ in ["__main__", "__builtin__", "builtins"]: main()