mirror of
https://github.com/mvt-project/mvt.git
synced 2024-06-23 12:58:58 +00:00
Pyment to reST
This commit is contained in:
parent
c38df37967
commit
169f5fbc26
|
@ -15,6 +15,8 @@ class IndicatorsFileBadFormat(Exception):
|
|||
class Indicators:
|
||||
"""This class is used to parse indicators from a STIX2 file and provide
|
||||
functions to compare extracted artifacts to the indicators.
|
||||
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, log=None):
|
||||
|
@ -37,6 +39,7 @@ class Indicators:
|
|||
|
||||
:param file_path: Path to the STIX2 file to parse
|
||||
:type file_path: str
|
||||
|
||||
"""
|
||||
self.log.info("Parsing STIX2 indicators file at path %s",
|
||||
file_path)
|
||||
|
@ -82,6 +85,7 @@ class Indicators:
|
|||
:type url: str
|
||||
:returns: True if the URL matched an indicator, otherwise False
|
||||
:rtype: bool
|
||||
|
||||
"""
|
||||
# TODO: If the IOC domain contains a subdomain, it is not currently
|
||||
# being matched.
|
||||
|
@ -153,6 +157,7 @@ class Indicators:
|
|||
:type urls: list
|
||||
:returns: True if any URL matched an indicator, otherwise False
|
||||
:rtype: bool
|
||||
|
||||
"""
|
||||
if not urls:
|
||||
return False
|
||||
|
@ -171,6 +176,7 @@ class Indicators:
|
|||
:type process: str
|
||||
:returns: True if process matched an indicator, otherwise False
|
||||
:rtype: bool
|
||||
|
||||
"""
|
||||
if not process:
|
||||
return False
|
||||
|
@ -196,6 +202,7 @@ class Indicators:
|
|||
:type processes: list
|
||||
:returns: True if process matched an indicator, otherwise False
|
||||
:rtype: bool
|
||||
|
||||
"""
|
||||
if not processes:
|
||||
return False
|
||||
|
@ -213,6 +220,7 @@ class Indicators:
|
|||
:type email: str
|
||||
:returns: True if email address matched an indicator, otherwise False
|
||||
:rtype: bool
|
||||
|
||||
"""
|
||||
if not email:
|
||||
return False
|
||||
|
@ -231,6 +239,7 @@ class Indicators:
|
|||
:type file_path: str
|
||||
:returns: True if the file path matched an indicator, otherwise False
|
||||
:rtype: bool
|
||||
|
||||
"""
|
||||
if not file_path:
|
||||
return False
|
||||
|
|
|
@ -23,8 +23,7 @@ class InsufficientPrivileges(Exception):
|
|||
pass
|
||||
|
||||
class MVTModule(object):
|
||||
"""This class provides a base for all extraction modules.
|
||||
"""
|
||||
"""This class provides a base for all extraction modules."""
|
||||
|
||||
enabled = True
|
||||
slug = None
|
||||
|
@ -66,8 +65,7 @@ class MVTModule(object):
|
|||
return cls(results=results, log=log)
|
||||
|
||||
def get_slug(self):
|
||||
"""Use the module's class name to retrieve a slug
|
||||
"""
|
||||
"""Use the module's class name to retrieve a slug"""
|
||||
if self.slug:
|
||||
return self.slug
|
||||
|
||||
|
@ -77,12 +75,13 @@ class MVTModule(object):
|
|||
def check_indicators(self):
|
||||
"""Check the results of this module against a provided list of
|
||||
indicators.
|
||||
|
||||
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def save_to_json(self):
|
||||
"""Save the collected results to a json file.
|
||||
"""
|
||||
"""Save the collected results to a json file."""
|
||||
if not self.output_folder:
|
||||
return
|
||||
|
||||
|
@ -112,6 +111,7 @@ class MVTModule(object):
|
|||
"""Serialize entry as JSON to deduplicate repeated entries
|
||||
|
||||
:param timeline: List of entries from timeline to deduplicate
|
||||
|
||||
"""
|
||||
timeline_set = set()
|
||||
for record in timeline:
|
||||
|
@ -141,8 +141,7 @@ class MVTModule(object):
|
|||
self.timeline_detected = self._deduplicate_timeline(self.timeline_detected)
|
||||
|
||||
def run(self):
|
||||
"""Run the main module procedure.
|
||||
"""
|
||||
"""Run the main module procedure."""
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
|
@ -190,6 +189,7 @@ def save_timeline(timeline, timeline_path):
|
|||
|
||||
:param timeline: List of records to order and store
|
||||
:param timeline_path: Path to the csv file to store the timeline to
|
||||
|
||||
"""
|
||||
with io.open(timeline_path, "a+", encoding="utf-8") as handle:
|
||||
csvoutput = csv.writer(handle, delimiter=",", quotechar="\"")
|
||||
|
|
|
@ -268,6 +268,7 @@ class URL:
|
|||
:type url: str
|
||||
:returns: Domain name extracted from URL
|
||||
:rtype: str
|
||||
|
||||
"""
|
||||
# TODO: Properly handle exception.
|
||||
try:
|
||||
|
@ -282,6 +283,7 @@ class URL:
|
|||
:type url: str
|
||||
:returns: Top-level domain name extracted from URL
|
||||
:rtype: str
|
||||
|
||||
"""
|
||||
# TODO: Properly handle exception.
|
||||
try:
|
||||
|
@ -292,8 +294,11 @@ class URL:
|
|||
def check_if_shortened(self) -> bool:
|
||||
"""Check if the URL is among list of shortener services.
|
||||
|
||||
|
||||
:returns: True if the URL is shortened, otherwise False
|
||||
|
||||
:rtype: bool
|
||||
|
||||
"""
|
||||
if self.domain.lower() in SHORTENER_DOMAINS:
|
||||
self.is_shortened = True
|
||||
|
@ -301,8 +306,7 @@ class URL:
|
|||
return self.is_shortened
|
||||
|
||||
def unshorten(self):
|
||||
"""Unshorten the URL by requesting an HTTP HEAD response.
|
||||
"""
|
||||
"""Unshorten the URL by requesting an HTTP HEAD response."""
|
||||
res = requests.head(self.url)
|
||||
if str(res.status_code).startswith("30"):
|
||||
return res.headers["Location"]
|
||||
|
|
|
@ -16,6 +16,7 @@ def convert_mactime_to_unix(timestamp, from_2001=True):
|
|||
:param from_2001: bool: Whether to (Default value = True)
|
||||
:param from_2001: Default value = True)
|
||||
:returns: Unix epoch timestamp.
|
||||
|
||||
"""
|
||||
if not timestamp:
|
||||
return None
|
||||
|
@ -42,6 +43,7 @@ def convert_chrometime_to_unix(timestamp):
|
|||
:param timestamp: Chrome timestamp as int.
|
||||
:type timestamp: int
|
||||
:returns: Unix epoch timestamp.
|
||||
|
||||
"""
|
||||
epoch_start = datetime.datetime(1601, 1 , 1)
|
||||
delta = datetime.timedelta(microseconds=timestamp)
|
||||
|
@ -55,6 +57,7 @@ def convert_timestamp_to_iso(timestamp):
|
|||
:type timestamp: int
|
||||
:returns: ISO timestamp string in YYYY-mm-dd HH:MM:SS.ms format.
|
||||
:rtype: str
|
||||
|
||||
"""
|
||||
try:
|
||||
return timestamp.strftime("%Y-%m-%d %H:%M:%S.%f")
|
||||
|
@ -67,6 +70,7 @@ def check_for_links(text):
|
|||
:param text: Any provided text.
|
||||
:type text: str
|
||||
:returns: Search results.
|
||||
|
||||
"""
|
||||
return re.findall("(?P<url>https?://[^\s]+)", text, re.IGNORECASE)
|
||||
|
||||
|
@ -92,6 +96,7 @@ def keys_bytes_to_string(obj):
|
|||
:param obj: Object to convert from bytes to string.
|
||||
:returns: Object converted to string.
|
||||
:rtype: str
|
||||
|
||||
"""
|
||||
new_obj = {}
|
||||
if not isinstance(obj, dict):
|
||||
|
|
Loading…
Reference in New Issue
Block a user