Browse Source

fix: Fixed pep8 in the new module and related libraries

pull/363/head
chrisr3d 2 years ago
parent
commit
70b3079aa3
No known key found for this signature in database GPG Key ID: 6BBED1B63A6D639F
  1. 8
      misp_modules/lib/vt_graph_parser/errors.py
  2. 115
      misp_modules/lib/vt_graph_parser/helpers/parsers.py
  3. 514
      misp_modules/lib/vt_graph_parser/helpers/rules.py
  4. 101
      misp_modules/lib/vt_graph_parser/helpers/wrappers.py
  5. 154
      misp_modules/lib/vt_graph_parser/importers/base.py
  6. 116
      misp_modules/lib/vt_graph_parser/importers/pymisp_response.py
  7. 124
      misp_modules/modules/export_mod/vt_graph.py

8
misp_modules/lib/vt_graph_parser/errors.py

@ -5,16 +5,16 @@ This module provides custom errors for data importers.
class GraphImportError(Exception):
pass
pass
class InvalidFileFormatError(Exception):
pass
pass
class MispEventNotFoundError(Exception):
pass
pass
class ServerError(Exception):
pass
pass

115
misp_modules/lib/vt_graph_parser/helpers/parsers.py

@ -26,64 +26,63 @@ VIRUSTOTAL_GRAPH_LINK_PREFIX = "https://www.virustotal.com/graph/"
def _parse_data(attributes, objects):
"""Parse MISP event attributes and objects data.
Args:
attributes (dict): dictionary which contains the MISP event attributes data.
objects (dict): dictionary which contains the MISP event objects data.
Returns:
([MispAttribute], str): MISP attributes and VTGraph link if exists.
Link defaults to "".
"""
attributes_data = []
vt_graph_link = ""
# Get simple MISP event attributes.
attributes_data += (
[attr for attr in attributes
if attr.get("type") in MISP_INPUT_ATTR])
# Get attributes from MISP objects too.
if objects:
for object_ in objects:
object_attrs = object_.get("Attribute", [])
attributes_data += (
[attr for attr in object_attrs
if attr.get("type") in MISP_INPUT_ATTR])
# Check if there is any VirusTotal Graph computed in MISP event.
vt_graph_links = (
attr for attr in attributes if attr.get("type") == "link"
and attr.get("value", "").startswith(VIRUSTOTAL_GRAPH_LINK_PREFIX))
# MISP could have more than one VirusTotal Graph, so we will take
# the last one.
current_id = 0 # MISP attribute id is the number of the attribute.
vt_graph_link = ""
for link in vt_graph_links:
if int(link.get("id")) > current_id:
current_id = int(link.get("id"))
vt_graph_link = link.get("value")
attributes = [
MispAttribute(data["type"], data["category"], data["value"])
for data in attributes_data]
return (attributes,
vt_graph_link.replace(VIRUSTOTAL_GRAPH_LINK_PREFIX, ""))
"""Parse MISP event attributes and objects data.
Args:
attributes (dict): dictionary which contains the MISP event attributes data.
objects (dict): dictionary which contains the MISP event objects data.
Returns:
([MispAttribute], str): MISP attributes and VTGraph link if exists.
Link defaults to "".
"""
attributes_data = []
vt_graph_link = ""
# Get simple MISP event attributes.
attributes_data += (
[attr for attr in attributes
if attr.get("type") in MISP_INPUT_ATTR])
# Get attributes from MISP objects too.
if objects:
for object_ in objects:
object_attrs = object_.get("Attribute", [])
attributes_data += (
[attr for attr in object_attrs
if attr.get("type") in MISP_INPUT_ATTR])
# Check if there is any VirusTotal Graph computed in MISP event.
vt_graph_links = (
attr for attr in attributes if attr.get("type") == "link"
and attr.get("value", "").startswith(VIRUSTOTAL_GRAPH_LINK_PREFIX))
# MISP could have more than one VirusTotal Graph, so we will take
# the last one.
current_id = 0 # MISP attribute id is the number of the attribute.
vt_graph_link = ""
for link in vt_graph_links:
if int(link.get("id")) > current_id:
current_id = int(link.get("id"))
vt_graph_link = link.get("value")
attributes = [
MispAttribute(data["type"], data["category"], data["value"])
for data in attributes_data]
return (attributes,
vt_graph_link.replace(VIRUSTOTAL_GRAPH_LINK_PREFIX, ""))
def parse_pymisp_response(payload):
"""Get event attributes and VirusTotal Graph id from pymisp response.
Args:
payload (dict): dictionary which contains pymisp response.
Returns:
([MispAttribute], str): MISP attributes and VTGraph link if exists.
Link defaults to "".
"""
event_attrs = payload.get("Attribute", [])
objects = payload.get("Object")
return _parse_data(event_attrs, objects)
"""Get event attributes and VirusTotal Graph id from pymisp response.
Args:
payload (dict): dictionary which contains pymisp response.
Returns:
([MispAttribute], str): MISP attributes and VTGraph link if exists.
Link defaults to "".
"""
event_attrs = payload.get("Attribute", [])
objects = payload.get("Object")
return _parse_data(event_attrs, objects)

514
misp_modules/lib/vt_graph_parser/helpers/rules.py

@ -15,290 +15,290 @@ import abc
class MispEventRule(object):
"""Rules for MISP event nodes connection object wrapper."""
def __init__(self, last_rule=None, node=None):
"""Create a MispEventRule instance.
MispEventRule is a collection of rules that can infer the relationships
between nodes from MISP events.
Args:
last_rule (MispEventRule): previous rule.
node (Node): actual node.
"""
self.last_rule = last_rule
self.node = node
self.relation_event = {
"ip_address": self.__ip_transition,
"url": self.__url_transition,
"domain": self.__domain_transition,
"file": self.__file_transition
}
def get_last_different_rule(self):
"""Search the last rule whose event was different from actual.
Returns:
MispEventRule: the last different rule.
"""
if not isinstance(self, self.last_rule.__class__):
return self.last_rule
else:
return self.last_rule.get_last_different_rule()
def resolve_relation(self, graph, node, misp_category):
"""Try to infer a relationship between two nodes.
This method is based on a non-deterministic finite automaton for
this reason the future rule only depends on the actual rule and the input
node.
For example if the actual rule is a MISPEventDomainRule and the given node
is an ip_address node, the connection type between them will be
`resolutions` and the this rule will transit to MISPEventIPRule.
Args:
graph (VTGraph): graph to be computed.
node (Node): the node to be linked.
misp_category: (str): MISP category of the given node.
Returns:
MispEventRule: the transited rule.
"""
if node.node_type in self.relation_event:
return self.relation_event[node.node_type](graph, node, misp_category)
else:
return self.manual_link(graph, node)
def manual_link(self, graph, node):
"""Creates a manual link between self.node and the given node.
We accept MISP types that VirusTotal does not know how to link, so we create
a end to end relationship instead of create an unknown relationship node.
Args:
graph (VTGraph): graph to be computed.
node (Node): the node to be linked.
Returns:
MispEventRule: the transited rule.
"""
graph.add_link(self.node.node_id, node.node_id, "manual")
return self
@abc.abstractmethod
def __file_transition(self, graph, node, misp_category):
"""Make a new transition due to file attribute event.
Args:
graph (VTGraph): graph to be computed.
node (Node): the node to be linked.
misp_category: (str): MISP category of the given node.
Returns:
MispEventRule: the transited rule.
"""
pass
@abc.abstractmethod
def __ip_transition(self, graph, node, misp_category):
"""Make a new transition due to ip attribute event.
Args:
graph (VTGraph): graph to be computed.
node (Node): the node to be linked.
misp_category: (str): MISP category of the given node.
Returns:
MispEventRule: the transited rule.
"""
pass
@abc.abstractmethod
def __url_transition(self, graph, node, misp_category):
"""Make a new transition due to url attribute event.
Args:
graph (VTGraph): graph to be computed.
node (Node): the node to be linked.
misp_category: (str): MISP category of the given node.
Returns:
MispEventRule: the transited rule.
"""
pass
@abc.abstractmethod
def __domain_transition(self, graph, node, misp_category):
"""Make a new transition due to domain attribute event.
Args:
graph (VTGraph): graph to be computed.
node (Node): the node to be linked.
misp_category: (str): MISP category of the given node.
Returns:
MispEventRule: the transited rule.
"""
pass
"""Rules for MISP event nodes connection object wrapper."""
def __init__(self, last_rule=None, node=None):
"""Create a MispEventRule instance.
MispEventRule is a collection of rules that can infer the relationships
between nodes from MISP events.
Args:
last_rule (MispEventRule): previous rule.
node (Node): actual node.
"""
self.last_rule = last_rule
self.node = node
self.relation_event = {
"ip_address": self.__ip_transition,
"url": self.__url_transition,
"domain": self.__domain_transition,
"file": self.__file_transition
}
def get_last_different_rule(self):
"""Search the last rule whose event was different from actual.
Returns:
MispEventRule: the last different rule.
"""
if not isinstance(self, self.last_rule.__class__):
return self.last_rule
else:
return self.last_rule.get_last_different_rule()
def resolve_relation(self, graph, node, misp_category):
"""Try to infer a relationship between two nodes.
This method is based on a non-deterministic finite automaton for
this reason the future rule only depends on the actual rule and the input
node.
For example if the actual rule is a MISPEventDomainRule and the given node
is an ip_address node, the connection type between them will be
`resolutions` and the this rule will transit to MISPEventIPRule.
Args:
graph (VTGraph): graph to be computed.
node (Node): the node to be linked.
misp_category: (str): MISP category of the given node.
Returns:
MispEventRule: the transited rule.
"""
if node.node_type in self.relation_event:
return self.relation_event[node.node_type](graph, node, misp_category)
else:
return self.manual_link(graph, node)
def manual_link(self, graph, node):
"""Creates a manual link between self.node and the given node.
We accept MISP types that VirusTotal does not know how to link, so we create
a end to end relationship instead of create an unknown relationship node.
Args:
graph (VTGraph): graph to be computed.
node (Node): the node to be linked.
Returns:
MispEventRule: the transited rule.
"""
graph.add_link(self.node.node_id, node.node_id, "manual")
return self
@abc.abstractmethod
def __file_transition(self, graph, node, misp_category):
"""Make a new transition due to file attribute event.
Args:
graph (VTGraph): graph to be computed.
node (Node): the node to be linked.
misp_category: (str): MISP category of the given node.
Returns:
MispEventRule: the transited rule.
"""
pass
@abc.abstractmethod
def __ip_transition(self, graph, node, misp_category):
"""Make a new transition due to ip attribute event.
Args:
graph (VTGraph): graph to be computed.
node (Node): the node to be linked.
misp_category: (str): MISP category of the given node.
Returns:
MispEventRule: the transited rule.
"""
pass
@abc.abstractmethod
def __url_transition(self, graph, node, misp_category):
"""Make a new transition due to url attribute event.
Args:
graph (VTGraph): graph to be computed.
node (Node): the node to be linked.
misp_category: (str): MISP category of the given node.
Returns:
MispEventRule: the transited rule.
"""
pass
@abc.abstractmethod
def __domain_transition(self, graph, node, misp_category):
"""Make a new transition due to domain attribute event.
Args:
graph (VTGraph): graph to be computed.
node (Node): the node to be linked.
misp_category: (str): MISP category of the given node.
Returns:
MispEventRule: the transited rule.
"""
pass
class MispEventURLRule(MispEventRule):
"""Rule for URL event."""
"""Rule for URL event."""
def __init__(self, last_rule=None, node=None):
super(MispEventURLRule, self).__init__(last_rule, node)
self.relation_event = {
"ip_address": self.__ip_transition,
"url": self.__url_transition,
"domain": self.__domain_transition,
"file": self.__file_transition
}
def __init__(self, last_rule=None, node=None):
super(MispEventURLRule, self).__init__(last_rule, node)
self.relation_event = {
"ip_address": self.__ip_transition,
"url": self.__url_transition,
"domain": self.__domain_transition,
"file": self.__file_transition
}
def __file_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "downloaded_files")
return MispEventFileRule(self, node)
def __file_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "downloaded_files")
return MispEventFileRule(self, node)
def __ip_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "contacted_ips")
return MispEventIPRule(self, node)
def __ip_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "contacted_ips")
return MispEventIPRule(self, node)
def __url_transition(self, graph, node, misp_category):
suitable_rule = self.get_last_different_rule()
if not isinstance(suitable_rule, MispEventInitialRule):
return suitable_rule.resolve_relation(graph, node, misp_category)
else:
return MispEventURLRule(self, node)
def __url_transition(self, graph, node, misp_category):
suitable_rule = self.get_last_different_rule()
if not isinstance(suitable_rule, MispEventInitialRule):
return suitable_rule.resolve_relation(graph, node, misp_category)
else:
return MispEventURLRule(self, node)
def __domain_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "contacted_domains")
return MispEventDomainRule(self, node)
def __domain_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "contacted_domains")
return MispEventDomainRule(self, node)
class MispEventIPRule(MispEventRule):
"""Rule for IP event."""
def __init__(self, last_rule=None, node=None):
super(MispEventIPRule, self).__init__(last_rule, node)
self.relation_event = {
"ip_address": self.__ip_transition,
"url": self.__url_transition,
"domain": self.__domain_transition,
"file": self.__file_transition
}
def __file_transition(self, graph, node, misp_category):
connection_type = "communicating_files"
if misp_category == "Artifacts dropped":
connection_type = "downloaded_files"
graph.add_link(self.node.node_id, node.node_id, connection_type)
return MispEventFileRule(self, node)
def __ip_transition(self, graph, node, misp_category):
suitable_rule = self.get_last_different_rule()
if not isinstance(suitable_rule, MispEventInitialRule):
return suitable_rule.resolve_relation(graph, node, misp_category)
else:
return MispEventIPRule(self, node)
def __url_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "urls")
return MispEventURLRule(self, node)
def __domain_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "resolutions")
return MispEventDomainRule(self, node)
"""Rule for IP event."""
def __init__(self, last_rule=None, node=None):
super(MispEventIPRule, self).__init__(last_rule, node)
self.relation_event = {
"ip_address": self.__ip_transition,
"url": self.__url_transition,
"domain": self.__domain_transition,
"file": self.__file_transition
}
def __file_transition(self, graph, node, misp_category):
connection_type = "communicating_files"
if misp_category == "Artifacts dropped":
connection_type = "downloaded_files"
graph.add_link(self.node.node_id, node.node_id, connection_type)
return MispEventFileRule(self, node)
def __ip_transition(self, graph, node, misp_category):
suitable_rule = self.get_last_different_rule()
if not isinstance(suitable_rule, MispEventInitialRule):
return suitable_rule.resolve_relation(graph, node, misp_category)
else:
return MispEventIPRule(self, node)
def __url_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "urls")
return MispEventURLRule(self, node)
def __domain_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "resolutions")
return MispEventDomainRule(self, node)
class MispEventDomainRule(MispEventRule):
"""Rule for domain event."""
def __init__(self, last_rule=None, node=None):
super(MispEventDomainRule, self).__init__(last_rule, node)
self.relation_event = {
"ip_address": self.__ip_transition,
"url": self.__url_transition,
"domain": self.__domain_transition,
"file": self.__file_transition
}
def __file_transition(self, graph, node, misp_category):
connection_type = "communicating_files"
if misp_category == "Artifacts dropped":
connection_type = "downloaded_files"
graph.add_link(self.node.node_id, node.node_id, connection_type)
return MispEventFileRule(self, node)
def __ip_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "resolutions")
return MispEventIPRule(self, node)
def __url_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "urls")
return MispEventURLRule(self, node)
def __domain_transition(self, graph, node, misp_category):
suitable_rule = self.get_last_different_rule()
if not isinstance(suitable_rule, MispEventInitialRule):
return suitable_rule.resolve_relation(graph, node, misp_category)
else:
graph.add_link(self.node.node_id, node.node_id, "siblings")
return MispEventDomainRule(self, node)
"""Rule for domain event."""
def __init__(self, last_rule=None, node=None):
super(MispEventDomainRule, self).__init__(last_rule, node)
self.relation_event = {
"ip_address": self.__ip_transition,
"url": self.__url_transition,
"domain": self.__domain_transition,
"file": self.__file_transition
}
def __file_transition(self, graph, node, misp_category):
connection_type = "communicating_files"
if misp_category == "Artifacts dropped":
connection_type = "downloaded_files"
graph.add_link(self.node.node_id, node.node_id, connection_type)
return MispEventFileRule(self, node)
def __ip_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "resolutions")
return MispEventIPRule(self, node)
def __url_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "urls")
return MispEventURLRule(self, node)
def __domain_transition(self, graph, node, misp_category):
suitable_rule = self.get_last_different_rule()
if not isinstance(suitable_rule, MispEventInitialRule):
return suitable_rule.resolve_relation(graph, node, misp_category)
else:
graph.add_link(self.node.node_id, node.node_id, "siblings")
return MispEventDomainRule(self, node)
class MispEventFileRule(MispEventRule):
"""Rule for File event."""
"""Rule for File event."""
def __init__(self, last_rule=None, node=None):
super(MispEventFileRule, self).__init__(last_rule, node)
self.relation_event = {
"ip_address": self.__ip_transition,
"url": self.__url_transition,
"domain": self.__domain_transition,
"file": self.__file_transition
}
def __init__(self, last_rule=None, node=None):
super(MispEventFileRule, self).__init__(last_rule, node)
self.relation_event = {
"ip_address": self.__ip_transition,
"url": self.__url_transition,
"domain": self.__domain_transition,
"file": self.__file_transition
}
def __file_transition(self, graph, node, misp_category):
suitable_rule = self.get_last_different_rule()
if not isinstance(suitable_rule, MispEventInitialRule):
return suitable_rule.resolve_relation(graph, node, misp_category)
else:
return MispEventFileRule(self, node)
def __file_transition(self, graph, node, misp_category):
suitable_rule = self.get_last_different_rule()
if not isinstance(suitable_rule, MispEventInitialRule):
return suitable_rule.resolve_relation(graph, node, misp_category)
else:
return MispEventFileRule(self, node)
def __ip_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "contacted_ips")
return MispEventIPRule(self, node)
def __ip_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "contacted_ips")
return MispEventIPRule(self, node)
def __url_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "contacted_urls")
return MispEventURLRule(self, node)
def __url_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "contacted_urls")
return MispEventURLRule(self, node)
def __domain_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "contacted_domains")
return MispEventDomainRule(self, node)
def __domain_transition(self, graph, node, misp_category):
graph.add_link(self.node.node_id, node.node_id, "contacted_domains")
return MispEventDomainRule(self, node)
class MispEventInitialRule(MispEventRule):
"""Initial rule."""
"""Initial rule."""
def __init__(self, last_rule=None, node=None):
super(MispEventInitialRule, self).__init__(last_rule, node)
self.relation_event = {
"ip_address": self.__ip_transition,
"url": self.__url_transition,
"domain": self.__domain_transition,
"file": self.__file_transition
}
def __init__(self, last_rule=None, node=None):
super(MispEventInitialRule, self).__init__(last_rule, node)
self.relation_event = {
"ip_address": self.__ip_transition,
"url": self.__url_transition,
"domain": self.__domain_transition,
"file": self.__file_transition
}
def __file_transition(self, graph, node, misp_category):
return MispEventFileRule(self, node)
def __file_transition(self, graph, node, misp_category):
return MispEventFileRule(self, node)
def __ip_transition(self, graph, node, misp_category):
return MispEventIPRule(self, node)
def __ip_transition(self, graph, node, misp_category):
return MispEventIPRule(self, node)
def __url_transition(self, graph, node, misp_category):
return MispEventURLRule(self, node)
def __url_transition(self, graph, node, misp_category):
return MispEventURLRule(self, node)
def __domain_transition(self, graph, node, misp_category):
return MispEventDomainRule(self, node)
def __domain_transition(self, graph, node, misp_category):
return MispEventDomainRule(self, node)

101
misp_modules/lib/vt_graph_parser/helpers/wrappers.py

@ -5,55 +5,54 @@ This module provides a Python object wrapper for MISP objects.
class MispAttribute(object):
"""Python object wrapper for MISP attribute.
Attributes:
type (str): VirusTotal node type.
category (str): MISP attribute category.
value (str): node id.
label (str): node name.
misp_type (str): MISP node type.
"""
MISP_TYPES_REFERENCE = {
"hostname": "domain",
"domain": "domain",
"ip-src": "ip_address",
"ip-dst": "ip_address",
"url": "url",
"filename|X": "file",
"filename": "file",
"md5": "file",
"sha1": "file",
"sha256": "file",
"target-user": "victim",
"target-email": "email"
}
def __init__(self, misp_type, category, value, label=""):
"""Constructor for a MispAttribute.
Args:
misp_type (str): MISP type attribute.
category (str): MISP category attribute.
value (str): attribute value.
label (str): attribute label.
"""Python object wrapper for MISP attribute.
Attributes:
type (str): VirusTotal node type.
category (str): MISP attribute category.
value (str): node id.
label (str): node name.
misp_type (str): MISP node type.
"""
if misp_type.startswith("filename|"):
label, value = value.split("|")
misp_type = "filename|X"
if misp_type == "filename":
label = value
self.type = self.MISP_TYPES_REFERENCE.get(misp_type)
self.category = category
self.value = value
self.label = label
self.misp_type = misp_type
def __eq__(self, other):
return (isinstance(other, self.__class__) and self.value == other.value and
self.type == other.type)
def __repr__(self):
return 'MispAttribute("{type}", "{category}", "{value}")'.format(type=self.type, category=self.category, value=self.value)
MISP_TYPES_REFERENCE = {
"hostname": "domain",
"domain": "domain",
"ip-src": "ip_address",
"ip-dst": "ip_address",
"url": "url",
"filename|X": "file",
"filename": "file",
"md5": "file",
"sha1": "file",
"sha256": "file",
"target-user": "victim",
"target-email": "email"
}
def __init__(self, misp_type, category, value, label=""):
"""Constructor for a MispAttribute.
Args:
misp_type (str): MISP type attribute.
category (str): MISP category attribute.
value (str): attribute value.
label (str): attribute label.
"""
if misp_type.startswith("filename|"):
label, value = value.split("|")
misp_type = "filename|X"
if misp_type == "filename":
label = value
self.type = self.MISP_TYPES_REFERENCE.get(misp_type)
self.category = category
self.value = value
self.label = label
self.misp_type = misp_type
def __eq__(self, other):
return (isinstance(other, self.__class__) and self.value == other.value and self.type == other.type)
def __repr__(self):
return 'MispAttribute("{type}", "{category}", "{value}")'.format(type=self.type, category=self.category, value=self.value)

154
misp_modules/lib/vt_graph_parser/importers/base.py

@ -9,90 +9,90 @@ from lib.vt_graph_parser.helpers.rules import MispEventInitialRule
def import_misp_graph(
misp_attributes, graph_id, vt_api_key, fetch_information, name,
private, fetch_vt_enterprise, user_editors, user_viewers, group_editors,
group_viewers, use_vt_to_connect_the_graph, max_api_quotas,
max_search_depth):
"""Import VirusTotal Graph from MISP.
misp_attributes, graph_id, vt_api_key, fetch_information, name,
private, fetch_vt_enterprise, user_editors, user_viewers, group_editors,
group_viewers, use_vt_to_connect_the_graph, max_api_quotas,
max_search_depth):
"""Import VirusTotal Graph from MISP.
Args:
misp_attributes ([MispAttribute]): list with the MISP attributes which
will be added to the returned graph.
graph_id: if supplied, the graph will be loaded instead of compute it again.
vt_api_key (str): VT API Key.
fetch_information (bool): whether the script will fetch
information for added nodes in VT. Defaults to True.
name (str): graph title. Defaults to "".
private (bool): True for private graphs. You need to have
Private Graph premium features enabled in your subscription. Defaults
to False.
fetch_vt_enterprise (bool, optional): if True, the graph will search any
available information using VirusTotal Intelligence for the node if there
is no normal information for it. Defaults to False.
user_editors ([str]): usernames that can edit the graph.
Defaults to None.
user_viewers ([str]): usernames that can view the graph.
Defaults to None.
group_editors ([str]): groups that can edit the graph.
Defaults to None.
group_viewers ([str]): groups that can view the graph.
Defaults to None.
use_vt_to_connect_the_graph (bool): if True, graph nodes will
be linked using VirusTotal API. Otherwise, the links will be generated
using production rules based on MISP attributes order. Defaults to
False.
max_api_quotas (int): maximum number of api quotas that could
be consumed to resolve graph using VirusTotal API. Defaults to 20000.
max_search_depth (int, optional): max search depth to explore
relationship between nodes when use_vt_to_connect_the_graph is True.
Defaults to 3.
Args:
misp_attributes ([MispAttribute]): list with the MISP attributes which
will be added to the returned graph.
graph_id: if supplied, the graph will be loaded instead of compute it again.
vt_api_key (str): VT API Key.
fetch_information (bool): whether the script will fetch
information for added nodes in VT. Defaults to True.
name (str): graph title. Defaults to "".
private (bool): True for private graphs. You need to have
Private Graph premium features enabled in your subscription. Defaults
to False.
fetch_vt_enterprise (bool, optional): if True, the graph will search any
available information using VirusTotal Intelligence for the node if there
is no normal information for it. Defaults to False.
user_editors ([str]): usernames that can edit the graph.
Defaults to None.
user_viewers ([str]): usernames that can view the graph.
Defaults to None.
group_editors ([str]): groups that can edit the graph.
Defaults to None.
group_viewers ([str]): groups that can view the graph.
Defaults to None.
use_vt_to_connect_the_graph (bool): if True, graph nodes will
be linked using VirusTotal API. Otherwise, the links will be generated
using production rules based on MISP attributes order. Defaults to
False.
max_api_quotas (int): maximum number of api quotas that could
be consumed to resolve graph using VirusTotal API. Defaults to 20000.
max_search_depth (int, optional): max search depth to explore
relationship between nodes when use_vt_to_connect_the_graph is True.
Defaults to 3.
If use_vt_to_connect_the_graph is True, it will take some time to compute
graph.
If use_vt_to_connect_the_graph is True, it will take some time to compute
graph.
Returns:
vt_graph_api.graph.VTGraph: the imported graph.
"""
Returns:
vt_graph_api.graph.VTGraph: the imported graph.
"""
rule = MispEventInitialRule()
rule = MispEventInitialRule()
# Check if the event has been already computed in VirusTotal Graph. Otherwise
# a new graph will be created.
if not graph_id:
graph = vt_graph_api.VTGraph(
api_key=vt_api_key, name=name, private=private,
user_editors=user_editors, user_viewers=user_viewers,
group_editors=group_editors, group_viewers=group_viewers)
else:
graph = vt_graph_api.VTGraph.load_graph(graph_id, vt_api_key)
# Check if the event has been already computed in VirusTotal Graph. Otherwise
# a new graph will be created.
if not graph_id:
graph = vt_graph_api.VTGraph(
api_key=vt_api_key, name=name, private=private,
user_editors=user_editors, user_viewers=user_viewers,
group_editors=group_editors, group_viewers=group_viewers)
else:
graph = vt_graph_api.VTGraph.load_graph(graph_id, vt_api_key)
attributes_to_add = [attr for attr in misp_attributes
if not graph.has_node(attr.value)]
attributes_to_add = [attr for attr in misp_attributes
if not graph.has_node(attr.value)]
total_expandable_attrs = max(sum(
1 for attr in attributes_to_add
if attr.type in vt_graph_api.Node.SUPPORTED_NODE_TYPES),
1)
total_expandable_attrs = max(sum(
1 for attr in attributes_to_add
if attr.type in vt_graph_api.Node.SUPPORTED_NODE_TYPES),
1)
max_quotas_per_search = max(int(max_api_quotas / total_expandable_attrs), 1)
max_quotas_per_search = max(
int(max_api_quotas / total_expandable_attrs), 1)
previous_node_id = ""
for attr in attributes_to_add:
# Add the current attr as node to the graph.
added_node = graph.add_node(
attr.value, attr.type, fetch_information, fetch_vt_enterprise,
attr.label)
# If use_vt_to_connect_the_grap is True the nodes will be connected using
# VT API.
if use_vt_to_connect_the_graph:
if (attr.type not in vt_graph_api.Node.SUPPORTED_NODE_TYPES and
previous_node_id):
graph.add_link(previous_node_id, attr.value, "manual")
else:
graph.connect_with_graph(
attr.value, max_quotas_per_search, max_search_depth,
fetch_info_collected_nodes=fetch_information)
else:
rule = rule.resolve_relation(graph, added_node, attr.category)
previous_node_id = ""
for attr in attributes_to_add:
# Add the current attr as node to the graph.
added_node = graph.add_node(
attr.value, attr.type, fetch_information, fetch_vt_enterprise,
attr.label)
# If use_vt_to_connect_the_grap is True the nodes will be connected using
# VT API.
if use_vt_to_connect_the_graph:
if (attr.type not in vt_graph_api.Node.SUPPORTED_NODE_TYPES and previous_node_id):
graph.add_link(previous_node_id, attr.value, "manual")
else:
graph.connect_with_graph(
attr.value, max_quotas_per_search, max_search_depth,
fetch_info_collected_nodes=fetch_information)
else:
rule = rule.resolve_relation(graph, added_node, attr.category)
return graph
return graph

116
misp_modules/lib/vt_graph_parser/importers/pymisp_response.py

@ -5,71 +5,69 @@ response payload giving by MISP API directly.
"""
import json
from lib.vt_graph_parser import errors
from lib.vt_graph_parser.helpers.parsers import parse_pymisp_response
from lib.vt_graph_parser.importers.base import import_misp_graph
def from_pymisp_response(
payload, vt_api_key, fetch_information=True,
private=False, fetch_vt_enterprise=False, user_editors=None,
user_viewers=None, group_editors=None, group_viewers=None,
use_vt_to_connect_the_graph=False, max_api_quotas=1000,
max_search_depth=3, expand_node_one_level=False):
"""Import VirusTotal Graph from MISP JSON file.
payload, vt_api_key, fetch_information=True,
private=False, fetch_vt_enterprise=False, user_editors=None,
user_viewers=None, group_editors=None, group_viewers=None,
use_vt_to_connect_the_graph=False, max_api_quotas=1000,
max_search_depth=3, expand_node_one_level=False):
"""Import VirusTotal Graph from MISP JSON file.
Args:
payload (dict): dictionary which contains the request payload.
vt_api_key (str): VT API Key.
fetch_information (bool, optional): whether the script will fetch
information for added nodes in VT. Defaults to True.
name (str, optional): graph title. Defaults to "".
private (bool, optional): True for private graphs. You need to have
Private Graph premium features enabled in your subscription. Defaults
to False.
fetch_vt_enterprise (bool, optional): if True, the graph will search any
available information using VirusTotal Intelligence for the node if there
is no normal information for it. Defaults to False.
user_editors ([str], optional): usernames that can edit the graph.
Defaults to None.
user_viewers ([str], optional): usernames that can view the graph.
Defaults to None.
group_editors ([str], optional): groups that can edit the graph.
Defaults to None.
group_viewers ([str], optional): groups that can view the graph.
Defaults to None.
use_vt_to_connect_the_graph (bool, optional): if True, graph nodes will
be linked using VirusTotal API. Otherwise, the links will be generated
using production rules based on MISP attributes order. Defaults to
False.
max_api_quotas (int, optional): maximum number of api quotas that could
be consumed to resolve graph using VirusTotal API. Defaults to 20000.
max_search_depth (int, optional): max search depth to explore
relationship between nodes when use_vt_to_connect_the_graph is True.
Defaults to 3.
expand_one_level (bool, optional): expand entire graph one level.
Defaults to False.
Args:
payload (dict): dictionary which contains the request payload.
vt_api_key (str): VT API Key.
fetch_information (bool, optional): whether the script will fetch
information for added nodes in VT. Defaults to True.
name (str, optional): graph title. Defaults to "".
private (bool, optional): True for private graphs. You need to have
Private Graph premium features enabled in your subscription. Defaults
to False.
fetch_vt_enterprise (bool, optional): if True, the graph will search any
available information using VirusTotal Intelligence for the node if there
is no normal information for it. Defaults to False.
user_editors ([str], optional): usernames that can edit the graph.
Defaults to None.
user_viewers ([str], optional): usernames that can view the graph.
Defaults to None.
group_editors ([str], optional): groups that can edit the graph.
Defaults to None.
group_viewers ([str], optional): groups that can view the graph.
Defaults to None.
use_vt_to_connect_the_graph (bool, optional): if True, graph nodes will
be linked using VirusTotal API. Otherwise, the links will be generated
using production rules based on MISP attributes order. Defaults to
False.
max_api_quotas (int, optional): maximum number of api quotas that could
be consumed to resolve graph using VirusTotal API. Defaults to 20000.
max_search_depth (int, optional): max search depth to explore
relationship between nodes when use_vt_to_connect_the_graph is True.
Defaults to 3.
expand_one_level (bool, optional): expand entire graph one level.
Defaults to False.
If use_vt_to_connect_the_graph is True, it will take some time to compute
graph.
If use_vt_to_connect_the_graph is True, it will take some time to compute
graph.
Raises:
LoaderError: if JSON file is invalid.
Raises:
LoaderError: if JSON file is invalid.
Returns:
[vt_graph_api.graph.VTGraph: the imported graph].
"""
graphs = []
for event_payload in payload['data']:
misp_attrs, graph_id = parse_pymisp_response(event_payload)
name = "Graph created from MISP event"
graph = import_misp_graph(
misp_attrs, graph_id, vt_api_key, fetch_information, name,
private, fetch_vt_enterprise, user_editors, user_viewers, group_editors,
group_viewers, use_vt_to_connect_the_graph, max_api_quotas,
max_search_depth)
if expand_node_one_level:
graph.expand_n_level(1)
graphs.append(graph)
return graphs
Returns:
[vt_graph_api.graph.VTGraph: the imported graph].
"""
graphs = []
for event_payload in payload['data']:
misp_attrs, graph_id = parse_pymisp_response(event_payload)
name = "Graph created from MISP event"
graph = import_misp_graph(
misp_attrs, graph_id, vt_api_key, fetch_information, name,
private, fetch_vt_enterprise, user_editors, user_viewers, group_editors,
group_viewers, use_vt_to_connect_the_graph, max_api_quotas,
max_search_depth)
if expand_node_one_level:
graph.expand_n_level(1)
graphs.append(graph)
return graphs

124
misp_modules/modules/export_mod/vt_graph.py

@ -43,71 +43,71 @@ moduleconfig = [
def handler(q=False):
"""Expansion handler.
Args:
q (bool, optional): module data. Defaults to False.
Returns:
[str]: VirusTotal graph links
"""
if not q:
return False
request = json.loads(q)
if not request.get('config') or not request['config'].get('vt_api_key'):
misperrors['error'] = 'A VirusTotal api key is required for this module.'
return misperrors
config = request['config']
api_key = config.get('vt_api_key')
fetch_information = config.get('fetch_information') or False
private = config.get('private') or False
fetch_vt_enterprise = config.get('fetch_vt_enterprise') or False
expand_one_level = config.get('expand_one_level') or False
user_editors = config.get('user_editors')
if user_editors:
user_editors = user_editors.split(',')
user_viewers = config.get('user_viewers')
if user_viewers:
user_viewers = user_viewers.split(',')
group_editors = config.get('group_editors')
if group_editors:
group_editors = group_editors.split(',')
group_viewers = config.get('group_viewers')
if group_viewers:
group_viewers = group_viewers.split(',')
graphs = from_pymisp_response(
request, api_key, fetch_information=fetch_information,
private=private, fetch_vt_enterprise=fetch_vt_enterprise,
user_editors=user_editors, user_viewers=user_viewers,
group_editors=group_editors, group_viewers=group_viewers,
expand_node_one_level=expand_one_level)
links = []
for graph in graphs:
graph.save_graph()
links.append(graph.get_ui_link())
# This file will contains one VirusTotal graph link for each exported event
file_data = str(base64.b64encode(bytes('\n'.join(links), 'utf-8')), 'utf-8')
return {'response': [], 'data': file_data}
"""Expansion handler.
Args:
q (bool, optional): module data. Defaults to False.
Returns:
[str]: VirusTotal graph links
"""
if not q:
return False
request = json.loads(q)
if not request.get('config') or not request['config'].get('vt_api_key'):
misperrors['error'] = 'A VirusTotal api key is required for this module.'
return misperrors
config = request['config']
api_key = config.get('vt_api_key')
fetch_information = config.get('fetch_information') or False
private = config.get('private') or False
fetch_vt_enterprise = config.get('fetch_vt_enterprise') or False
expand_one_level = config.get('expand_one_level') or False
user_editors = config.get('user_editors')
if user_editors:
user_editors = user_editors.split(',')
user_viewers = config.get('user_viewers')
if user_viewers:
user_viewers = user_viewers.split(',')
group_editors = config.get('group_editors')
if group_editors:
group_editors = group_editors.split(',')
group_viewers = config.get('group_viewers')
if group_viewers:
group_viewers = group_viewers.split(',')
graphs = from_pymisp_response(
request, api_key, fetch_information=fetch_information,
private=private, fetch_vt_enterprise=fetch_vt_enterprise,
user_editors=user_editors, user_viewers=user_viewers,
group_editors=group_editors, group_viewers=group_viewers,
expand_node_one_level=expand_one_level)
links = []
for graph in graphs:
graph.save_graph()
links.append(graph.get_ui_link())
# This file will contains one VirusTotal graph link for each exported event
file_data = str(base64.b64encode(
bytes('\n'.join(links), 'utf-8')), 'utf-8')
return {'response': [], 'data': file_data}
def introspection():
modulesetup = {
'responseType': 'application/txt',
'outputFileExtension': 'txt',
'userConfig': {},
'inputSource': []
}
return modulesetup
modulesetup = {
'responseType': 'application/txt',
'outputFileExtension': 'txt',
'userConfig': {},
'inputSource': []
}
return modulesetup
def version():
moduleinfo['config'] = moduleconfig
return moduleinfo
moduleinfo['config'] = moduleconfig
return moduleinfo

Loading…
Cancel
Save