Scripting

Search…
Scripting
How to use the Scripting API
What is Mythic scripting?
The main Mythic server Docker container is based around WebSockets and RESTful API endpoints. We can hit the same RESTful endpoints and listen to the same WebSocket endpoints that the main browser user interface uses as part of scripting, which means scripting can technically be done in any language.
Where is it?
Install the PyPi package via pip pip3 install mythic . The current mythic package is version 0.0.23 and reports to Mythic as scripting version "3". The code for it is public - https://github.com/MythicMeta/Mythic_Scripting​
The RESTful interfaces that mythic currently uses are located under mythic_rest, so do the following:
1
from mythic import mythic_rest
2
mythic = mythic_rest.Mythic(username="mythic_admin", 
3
                    password="mythic_password",
4
                    server_ip="192.168.205.151", 
5
                    server_port="7443", 
6
                    ssl=True, 
7
                    global_timeout=-1)
Copied!
The PyPI package specifically refers to mythic_rest because there are likely to be other interfaces in the future (such as graphql).
Scripting base
The scripting base is pretty simple:
1
from mythic import mythic_rest
2
from sys import exit
3
​
4
​
5
async def scripting():
6
    # sample login
7
    mythic = mythic_rest.Mythic(username="mythic_admin", 
8
                    password="mythic_password",
9
                    server_ip="192.168.205.151", 
10
                    server_port="7443", 
11
                    ssl=True, 
12
                    global_timeout=-1)
13
    await mythic.login()
14
    # either set an api token to use or create a new one to start using
15
    resp = await mythic.set_or_create_apitoken()
16
    await mythic_rest.json_print(resp)
17
    
18
async def main():
19
    await scripting()
20
    try:
21
        while True:
22
            pending = asyncio.Task.all_tasks()
23
            if len(pending) == 0:
24
                exit(0)
25
            else:
26
                await asyncio.gather(*pending)
27
    except KeyboardInterrupt:
28
        pending = asyncio.Task.all_tasks()
29
        for t in pending:
30
            t.cancel()
31
loop = asyncio.get_event_loop()
32
loop.run_until_complete(main())
Copied!
Line 1 imports all of the functions from the mythic api into the script. From here, there's two functions we write:
1.main() - this function calls the scripting function asynchronously ( await scripting() ), then loops through all functions sitting on the current event loop until everything is done. This is what allows you to eventually hook into the WebSocket eventing and wait for notifications.
2.scripting() - this is the initial function where you write your code.
In the scripting function, we do a few things:
1.The first thing is to create an instance of the Mythic server. This includes the credentials we use to log in and information about the server itself. The global_timeout is an optional parameter to globally provide timeouts if no others are provided. If the global_timeout is set to anything less than 0, then the program will wait indefinitely.
2.Once this object is created, the next thing is to log into mythic with await mythic.login(). This sends the credentials over the connection and gets back the standard JWT access_token and refresh_tokens.
3.The next standard thing to do is to do await mythic.set_or_create_apitoken(). Because dealing with JWT access tokens is annoying (they have short timestamps) and trying to make sure you properly deal with refresh tokens in scripting is error prone, there's a helper function to create a user level API token. These API tokens show up in your settings page and can be deactivated or deleted. The advantage of these tokens is that while they're active, they don't expire. This is very useful for long running tasks. Doing set_or_create_apitoken() is a helper function to get and potentially create a user-level API token and set it on your current mythic instance.
If you've already created an API token that you'd like to use instead of supplying your username and password, you can create the mythic object with the following:
1
mythic = mythic_rest.Mythic(server_ip="192.168.205.151", server_port="7443", ssl=True,
2
                apitoken="ej..api token here")
Copied!
Because you already have the apitoken set, there's no need for additional calls to login or set_or_create_apitoken.
Scripting Principals
The Mythic scripting file tries to make everything into classes/objects so that it's easier to use with an IDE than just python dictionaries everywhere. This makes it closely resemble the Mythic database (although there's a lot more to the database than what's captured in the scripting so far).
Everything call to mythic that doesn't start with listen will return a MythicResponse class object. This allows you to properly inspect the .status of the query ( success or error) as well as the .response_code (the web response code from the query). The actual response object(s) is in the .response component. For example:
1
resp = await mythic.get_all_tasks() # this returns a MythicResponse object
2
# resp.response is an array of Task objects
3
await mythic_rest.json_print(resp) # print out the MythicResponse object as JSON
4
for t in resp.response:
5
    await mythic_rest.json_print(x) # print the Task object as JSON
Copied!
the json_print function allows you to easily print Mythic class objects as JSON. If you just want to access it as JSON data (i.e. not printing it), then each object has a .to_json() function.
Listening for Notifications
So far, the components have been doing RESTful API calls directly. But one of the more interesting aspects of scripting is to listen for specific notification events and reacting to them. In Mythic, these are all indicated with function names starting with listen_for. These functions all have the same general format:
1
async def listen_for_new_callbacks(self, callback_function=None, timeout=None):
2
        """
3
        Uses websockets to listen for all notifications related new callbacks.
4
        To stop listening, call cancel() on the result from this function call
5
        :param callback_function: gets called on each notification
6
        :return:
7
        """
8
        url = "{}{}:{}/ws/new_callbacks/current_operation".format(self._ws, self._server_ip, self._server_port)
9
        if callback_function:
10
            task = await self.stream_output(url, callback_function, timeout)
11
        else:
12
            task = await self.stream_output(url, self.print_websocket_output, timeout)
13
        return task
Copied!
If a callback_function isn't supplied, then the resulting data is simply printed to the screen. The timeout can be used for indicating how long you want to listen. Let's see how this can be used:
1
from mythic import mythic_rest
2
from sys import exit
3
import asyncio
4
​
5
​
6
async def scripting():
7
    # sample login
8
    mythic = mythic_rest.Mythic(username="mythic_admin", password="mythic_password",
9
                    server_ip="192.168.205.151", server_port="7443", ssl=True, global_timeout=-1)
10
    await mythic.login()
11
    # either set an api token to use or create a new one to start using
12
    resp = await mythic.set_or_create_apitoken()
13
    await mythic.listen_for_new_callbacks(issue_shell_whoami)
14
​
15
async def issue_shell_whoami(mythic, callback):
16
    try:
17
        print("in issue_shell_whoami, about to create a task")
18
        task = mythic_rest.Task(callback=callback, command=mythic_rest.Command(cmd="shell"), params="whoami")
19
        submit = await mythic.create_task(task, return_on="submitted")
20
        await mythic_rest.json_print(submit)
21
        print("task is submitted, now to wait for responses to process")
22
        results = await mythic.gather_task_responses(submit.id, timeout=20)
23
        print("got array of results of length: " + str(len(results)))
24
    except Exception as e:
25
        print(str(e))
26
​
27
async def main():
28
    await scripting()
29
    try:
30
        while True:
31
            pending = asyncio.Task.all_tasks()
32
            if len(pending) == 0:
33
                exit(0)
34
            else:
35
                await asyncio.gather(*pending)
36
    except KeyboardInterrupt:
37
        pending = asyncio.Task.all_tasks()
38
        for t in pending:
39
            t.cancel()
40
loop = asyncio.get_event_loop()
41
loop.run_until_complete(main())
Copied!
This example waits for new callbacks, and when a new one is recognized by Mythic, the issue_shell_whoami function defined on lines 14-25 is executed. These callback functions always have 2 parameters - the mythic instance associated with the data and a STRING representation of the data. To get back to a python dictionary, run json.loads(data). At this point, you can either deal with data as a dictionary directly (as shown in the above code), or you can cast it back to the right object. In the above example, we could do callback = Callback(**json.loads(data)) which would result in callback being a Callback object instead of a python dictionary.
Troubleshooting
When scripting, you might run into issues, so it's important to know how to troubleshoot what's going on. Every function returns a MythicResponse object. All Mythic specific objects (Responses, Tasks, Operators, etc) can be printed with await mythic_rest.json_print(object here). This provides a clean way to see the output that you actually got back. The reason everything returns a MythicResponse instead of just the object is so that we can capture HTTP response codes, raw output, and the Object casted types. This allows you to more easily manipulate and see data you get back via scripting.
Reports
Scripting Examples
Last modified 7mo ago
Copy link
Edit on GitHub
Contents
What is Mythic scripting?
Where is it?
Scripting base
Scripting Principals
Listening for Notifications
Troubleshooting