-
Notifications
You must be signed in to change notification settings - Fork 33
Expand file tree
/
Copy pathutils.py
More file actions
153 lines (131 loc) · 5.13 KB
/
utils.py
File metadata and controls
153 lines (131 loc) · 5.13 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
from datetime import datetime
import dateparser
import re
def get_domain_age(create_date):
"""
Finds how many days old a domain is given a start date.
Args:
create_date: Date in the form of %Y-%m-%d'
Returns: Number of days between now and the create_date.
"""
time_diff = datetime.now() - dateparser.parse(create_date, date_formats=['%Y-%m-%d', '%Y%m%d'])
return time_diff.days
def get_threat_component(components, threat_type):
"""
Gets a certain threat component out a list of components
Args:
components: List of threat components
threat_type: Type of threat we are looking for
Returns: Either the component that we asked for or None
"""
for component in components:
if component.get("name") == threat_type:
return component
else:
return None
def get_average_risk_score(domains):
"""
Gets average domain risk score for Investigate and Detect result sets
Args:
domains: Investigate or Detect result set
Returns: average risk score
"""
count = 0
total = 0
for d in domains:
# investigate result set
if "risk_score" in d.get("domain_risk", {}):
count += 1
total += d.get("domain_risk").get("risk_score")
# detect result set
elif d.get("risk_score"):
count += 1
total += d.get("risk_score")
return total // count if count else None
def get_average_age(domains):
"""
Gets average domain age for Investigate and Detect result sets
Args:
domains: Investigate or Detect result set
Returns: average age
"""
count = 0
total = 0
for d in domains:
if isinstance(d.get("create_date"), dict) and d.get("create_date").get("value"):
count += 1
total += get_domain_age(d.get("create_date").get("value"))
elif isinstance(d.get("create_date"), int):
count += 1
total += get_domain_age(str(d.get("create_date")))
return total // count if count else None
def prune_data(data_obj):
"""
Does a deep dive through a data object to prune any null or empty items. Checks for empty lists, dicts, and strs.
Args:
data_obj: Either a list or dict that needs to be pruned
"""
items_to_prune = []
if isinstance(data_obj, dict) and len(data_obj):
for k, v in data_obj.items():
if isinstance(data_obj[k], dict) or isinstance(data_obj[k], list):
prune_data(data_obj[k])
if not isinstance(v, int) and not v:
items_to_prune.append(k)
elif k == 'count' and v == 0:
items_to_prune.append(k)
for k in items_to_prune:
del data_obj[k]
elif isinstance(data_obj, list) and len(data_obj):
for index, item in enumerate(data_obj):
prune_data(item)
if not isinstance(item, int) and not item:
items_to_prune.append(index)
data_obj[:] = [item for index, item in enumerate(data_obj) if index not in items_to_prune and len(item)]
def find_emails(data_str):
"""Find and returns all emails"""
return set(re.findall(r'[\w\.-]+@[\w\.-]+', data_str))
def find_ips(data_str):
"""Find and returns all ipv4"""
ipv4s = set(re.findall(r'\b(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\b', data_str))
return ipv4s
def get_pivots(data_obj, name, return_data=None, count=0, pivot_threshold=500):
"""
Does a deep dive through a data object to check count vs pivot threshold.
Args:
data_obj: Either a list or dict that needs to check pivot count
name: pivot category name
return_data: Holds data to return once we reach the end of the data_obj
count: Lets us track to know when we are finished with the data_obj
pivot_threshold: Threshold to include as a pivot.
"""
if return_data is None:
return_data = []
count += 1
if isinstance(data_obj, dict) and len(data_obj):
temp_name = name
for k, v in data_obj.items():
if isinstance(data_obj[k], (dict, list)):
name = "{}_{}".format(name, k)
temp_data = get_pivots(
data_obj[k], name, return_data, count, pivot_threshold
)
if temp_data:
return_data.append([name[1:].upper().replace("_", " "), temp_data])
name = temp_name
if "count" in data_obj and (1 < data_obj["count"] < pivot_threshold):
return data_obj["value"], data_obj["count"]
elif isinstance(data_obj, list) and len(data_obj):
for index, item in enumerate(data_obj):
temp_data = get_pivots(item, name, return_data, count, pivot_threshold)
if temp_data:
if isinstance(temp_data, list):
for x in temp_data:
return_data.append(x)
elif isinstance(temp_data, tuple):
return_data.append([name[1:].upper().replace("_", " "), temp_data])
count -= 1
if count:
return
else:
return return_data