archive
/
PurpleDome
mirror of https://github.com/avast/PurpleDome
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

More testability by re-structuring log

Browse Source
pull/10/head
Thorsten Sick 3 years ago
parent 4746d77f9c
commit c72ab031bb
1 changed files with 46 additions and 38 deletions
Show Stats Download Patch File Download Diff File

84
app/attack_log.py
Unescape Escape View File

@ -6,10 +6,6 @@ import json
import datetime import datetime
def __get_timestamp__():
return datetime.datetime.now().strftime("%H:%M:%S.%f")
def __mitre_fix_ttp__(ttp): def __mitre_fix_ttp__(ttp):
""" enforce some systematic naming scheme for MITRE TTPs """ """ enforce some systematic naming scheme for MITRE TTPs """
@ -18,8 +14,8 @@ def __mitre_fix_ttp__(ttp):
if ttp.startswith("MITRE_"): if ttp.startswith("MITRE_"):
return ttp return ttp
else:
return "MITRE_" + ttp return "MITRE_" + ttp
class AttackLog(): class AttackLog():
@ -33,6 +29,22 @@ class AttackLog():
self.log = [] self.log = []
self.verbosity = verbosity self.verbosity = verbosity
# TODO. As soon as someone wants custom timestamps, make the format variable
self.datetime_format = "%H:%M:%S.%f"
def __add_to_log__(self, item: dict):
""" internal command to add a item to the log
@param item: data chunk to add
"""
self.log.append(item)
def __get_timestamp__(self):
""" Get the timestamp to add to the log entries. Currently not configurable """
return datetime.datetime.now().strftime(self.datetime_format)
def start_caldera_attack(self, source, paw, group, ability_id, ttp=None, name=None, description=None, obfuscator="default", jitter="default"): # pylint: disable=too-many-arguments def start_caldera_attack(self, source, paw, group, ability_id, ttp=None, name=None, description=None, obfuscator="default", jitter="default"): # pylint: disable=too-many-arguments
""" Mark the start of a caldera attack """ Mark the start of a caldera attack
@ -47,7 +59,7 @@ class AttackLog():
@param jitter: Jitter being used @param jitter: Jitter being used
""" """
data = {"timestamp": __get_timestamp__(), data = {"timestamp": self.__get_timestamp__(),
"event": "start", "event": "start",
"type": "attack", "type": "attack",
"sub-type": "caldera", "sub-type": "caldera",
@ -62,7 +74,7 @@ class AttackLog():
"jitter": jitter "jitter": jitter
} }
self.log.append(data) self.__add_to_log__(data)
# TODO: Add parameter # TODO: Add parameter
# TODO: Add config # TODO: Add config
@ -82,7 +94,7 @@ class AttackLog():
@param jitter: Jitter being used @param jitter: Jitter being used
""" """
data = {"timestamp": __get_timestamp__(), data = {"timestamp": self.__get_timestamp__(),
"event": "stop", "event": "stop",
"type": "attack", "type": "attack",
"sub-type": "caldera", "sub-type": "caldera",
@ -96,18 +108,17 @@ class AttackLog():
"obfuscator": obfuscator, "obfuscator": obfuscator,
"jitter": jitter "jitter": jitter
} }
self.log.append(data) self.__add_to_log__(data)
def start_file_write(self, source, target, file_name, ttp=None): def start_file_write(self, source, target, file_name):
""" Mark the start of a file being written to the target (payload !) """ Mark the start of a file being written to the target (payload !)
@param source: source of the attack. Attack IP (empty if written from controller) @param source: source of the attack. Attack IP (empty if written from controller)
@param target: Target machine of the attack @param target: Target machine of the attack
@param file_name: Name of the file being written @param file_name: Name of the file being written
@param ttp: TTP of the attack. From plugin
""" """
data = {"timestamp": __get_timestamp__(), data = {"timestamp": self.__get_timestamp__(),
"event": "start", "event": "start",
"type": "dropping_file", "type": "dropping_file",
"sub-type": "by PurpleDome", "sub-type": "by PurpleDome",
@ -115,19 +126,18 @@ class AttackLog():
"target": target, "target": target,
"file_name": file_name "file_name": file_name
} }
self.log.append(data) self.__add_to_log__(data)
def stop_file_write(self, source, target, file_name, ttp=None): def stop_file_write(self, source, target, file_name):
""" Mark the stop of a file being written to the target (payload !) """ Mark the stop of a file being written to the target (payload !)
@param source: source of the attack. Attack IP (empty if written from controller) @param source: source of the attack. Attack IP (empty if written from controller)
@param target: Target machine of the attack @param target: Target machine of the attack
@param attack_name: Name of the attack. From plugin @param attack_name: Name of the attack. From plugin
@param file_name: Name of the file being written @param file_name: Name of the file being written
@param ttp: TTP of the attack. From plugin
""" """
data = {"timestamp": __get_timestamp__(), data = {"timestamp": self.__get_timestamp__(),
"event": "stop", "event": "stop",
"type": "dropping_file", "type": "dropping_file",
"sub-type": "by PurpleDome", "sub-type": "by PurpleDome",
@ -135,18 +145,17 @@ class AttackLog():
"target": target, "target": target,
"file_name": file_name "file_name": file_name
} }
self.log.append(data) self.__add_to_log__(data)
def start_execute_payload(self, source, target, command, ttp=None): def start_execute_payload(self, source, target, command):
""" Mark the start of a payload being executed """ Mark the start of a payload being executed
@param source: source of the attack. Attack IP (empty if written from controller) @param source: source of the attack. Attack IP (empty if written from controller)
@param target: Target machine of the attack @param target: Target machine of the attack
@param command: Name of the file being written @param command: Name of the file being written
@param ttp: TTP of the attack. From plugin
""" """
data = {"timestamp": __get_timestamp__(), data = {"timestamp": self.__get_timestamp__(),
"event": "start", "event": "start",
"type": "execute_payload", "type": "execute_payload",
"sub-type": "by PurpleDome", "sub-type": "by PurpleDome",
@ -154,19 +163,18 @@ class AttackLog():
"target": target, "target": target,
"command": command "command": command
} }
self.log.append(data) self.__add_to_log__(data)
def stop_execute_payload(self, source, target, command, ttp=None): def stop_execute_payload(self, source, target, command):
""" Mark the stop of a payload being executed """ Mark the stop of a payload being executed
@param source: source of the attack. Attack IP (empty if written from controller) @param source: source of the attack. Attack IP (empty if written from controller)
@param target: Target machine of the attack @param target: Target machine of the attack
@param command: Name of the attack. From plugin @param command: Name of the attack. From plugin
@param file_name: Name of the file being written @param file_name: Name of the file being written
@param ttp: TTP of the attack. From plugin
""" """
data = {"timestamp": __get_timestamp__(), data = {"timestamp": self.__get_timestamp__(),
"event": "stop", "event": "stop",
"type": "execute_payload", "type": "execute_payload",
"sub-type": "by PurpleDome", "sub-type": "by PurpleDome",
@ -174,7 +182,7 @@ class AttackLog():
"target": target, "target": target,
"command": command "command": command
} }
self.log.append(data) self.__add_to_log__(data)
def start_kali_attack(self, source, target, attack_name, ttp=None): def start_kali_attack(self, source, target, attack_name, ttp=None):
""" Mark the start of a Kali based attack """ Mark the start of a Kali based attack
@ -185,7 +193,7 @@ class AttackLog():
@param ttp: TTP of the attack. From plugin @param ttp: TTP of the attack. From plugin
""" """
data = {"timestamp": __get_timestamp__(), data = {"timestamp": self.__get_timestamp__(),
"event": "start", "event": "start",
"type": "attack", "type": "attack",
"sub-type": "kali", "sub-type": "kali",
@ -194,7 +202,7 @@ class AttackLog():
"kali_name": attack_name, "kali_name": attack_name,
"hunting_tag": __mitre_fix_ttp__(ttp), "hunting_tag": __mitre_fix_ttp__(ttp),
} }
self.log.append(data) self.__add_to_log__(data)
# TODO: Add parameter # TODO: Add parameter
# TODO: Add config # TODO: Add config
@ -209,7 +217,7 @@ class AttackLog():
@param ttp: TTP of the attack. From plugin @param ttp: TTP of the attack. From plugin
""" """
data = {"timestamp": __get_timestamp__(), data = {"timestamp": self.__get_timestamp__(),
"event": "stop", "event": "stop",
"type": "attack", "type": "attack",
"sub-type": "kali", "sub-type": "kali",
@ -218,7 +226,7 @@ class AttackLog():
"kali_name": attack_name, "kali_name": attack_name,
"hunting_tag": __mitre_fix_ttp__(ttp), "hunting_tag": __mitre_fix_ttp__(ttp),
} }
self.log.append(data) self.__add_to_log__(data)
def start_metasploit_attack(self, source, target, metasploit_command, ttp=None): def start_metasploit_attack(self, source, target, metasploit_command, ttp=None):
""" Mark the start of a Metasploit based attack """ Mark the start of a Metasploit based attack
@ -229,7 +237,7 @@ class AttackLog():
@param ttp: TTP of the attack. From plugin @param ttp: TTP of the attack. From plugin
""" """
data = {"timestamp": __get_timestamp__(), data = {"timestamp": self.__get_timestamp__(),
"event": "start", "event": "start",
"type": "attack", "type": "attack",
"sub-type": "metasploit", "sub-type": "metasploit",
@ -238,7 +246,7 @@ class AttackLog():
"metasploit_command": metasploit_command, "metasploit_command": metasploit_command,
"hunting_tag": __mitre_fix_ttp__(ttp), "hunting_tag": __mitre_fix_ttp__(ttp),
} }
self.log.append(data) self.__add_to_log__(data)
def stop_metasploit_attack(self, source, target, metasploit_command, ttp=None): def stop_metasploit_attack(self, source, target, metasploit_command, ttp=None):
""" Mark the start of a Metasploit based attack """ Mark the start of a Metasploit based attack
@ -249,7 +257,7 @@ class AttackLog():
@param ttp: TTP of the attack. From plugin @param ttp: TTP of the attack. From plugin
""" """
data = {"timestamp": __get_timestamp__(), data = {"timestamp": self.__get_timestamp__(),
"event": "stop", "event": "stop",
"type": "attack", "type": "attack",
"sub-type": "metasploit", "sub-type": "metasploit",
@ -258,7 +266,7 @@ class AttackLog():
"metasploit_command": metasploit_command, "metasploit_command": metasploit_command,
"hunting_tag": __mitre_fix_ttp__(ttp), "hunting_tag": __mitre_fix_ttp__(ttp),
} }
self.log.append(data) self.__add_to_log__(data)
def start_attack_plugin(self, source, target, plugin_name, ttp=None): def start_attack_plugin(self, source, target, plugin_name, ttp=None):
""" Mark the start of an attack plugin """ Mark the start of an attack plugin
@ -269,7 +277,7 @@ class AttackLog():
@param ttp: TTP of the attack. From plugin @param ttp: TTP of the attack. From plugin
""" """
data = {"timestamp": __get_timestamp__(), data = {"timestamp": self.__get_timestamp__(),
"event": "start", "event": "start",
"type": "attack", "type": "attack",
"sub-type": "attack_plugin", "sub-type": "attack_plugin",
@ -278,7 +286,7 @@ class AttackLog():
"plugin_name": plugin_name, "plugin_name": plugin_name,
"hunting_tag": __mitre_fix_ttp__(ttp), "hunting_tag": __mitre_fix_ttp__(ttp),
} }
self.log.append(data) self.__add_to_log__(data)
# TODO: Add parameter # TODO: Add parameter
# TODO: Add config # TODO: Add config
@ -293,7 +301,7 @@ class AttackLog():
@param ttp: TTP of the attack. From plugin @param ttp: TTP of the attack. From plugin
""" """
data = {"timestamp": __get_timestamp__(), data = {"timestamp": self.__get_timestamp__(),
"event": "stop", "event": "stop",
"type": "attack", "type": "attack",
"sub-type": "attack_plugin", "sub-type": "attack_plugin",
@ -302,7 +310,7 @@ class AttackLog():
"plugin_name": plugin_name, "plugin_name": plugin_name,
"hunting_tag": __mitre_fix_ttp__(ttp), "hunting_tag": __mitre_fix_ttp__(ttp),
} }
self.log.append(data) self.__add_to_log__(data)
def write_json(self, filename): def write_json(self, filename):
""" Write the json data for this log """ Write the json data for this log

Write Preview
Loading…
Cancel
Save