Skip to main content

Remote execution via stdin/stdout messaging.

Project description PyPI - Python Version PyPI Read the Docs



A proof-of-concept for asynchronous adhoc remote procedure calls in Python.

This is work in progress and serves basically as an exercise.


  • Python >= 3.5 asyncio
  • adhoc transferable remote procedures
  • remote part of a implant.core.Command may reside in a separate module
  • a implant.core.Command specific implant.core.Channel enables arbitrary protocols between local and remote side
  • events
  • quite small core
  • tests


  • Python >= 3.5
  • only pure Python modules are supported for remote import, if no venv is used
  • implant.core.Command s must reside in a module other then __main__
  • at the moment sudo must not ask for password


General application

import asyncio
import pathlib

from implant import core, connect, commands

async def remote_tasks():
    # create a connector for a python process
    connector = connect.Lxd(
    connector_args = {
        'python_bin': pathlib.Path('/usr/bin/python3')
    # connect to a remote python process
    remote = await connector.launch(**connector_args)

    # start remote communication tasks
    com_remote = asyncio.ensure_future(remote.communicate())
        # execute command
        cmd = commands.SystemLoad()
        result = await remote.execute(cmd)

        print("Remote system load:", result)

        # stop communication tasks
        await com_remote

if __name__ == '__main__':
    loop = asyncio.get_event_loop()

An example Echo Command

import logging
import os

from implant import core

log = logging.getLogger(__name__)

class Echo(core.Command):

    """Demonstrate the basic command API."""

    async def local(self, context):
        """The local side of the RPC.

           :param context: :py:obj:`implant.core.DispatchLocalContext`
        # custom protocol
        # first: send
        await"send to remote")

        # second: receive
        from_remote = []
        async for x in
        log.debug("************ receiving from remote: %s", from_remote)

        # third: wait for remote to finish and return result
        remote_result = await context.remote_future

        result = {
            'from_remote': ''.join(from_remote),
        return result

    async def remote(self, context):
        """The remote side of the RPC.

           :param context: :py:obj:`implant.core.DispatchRemoteContext`
        # first: receive
        from_local = []
        async for x in
        log.debug("************ receiving from local: %s", from_local)

        # second: send
        await"send to local")

        # third: return result
        return {
            'from_local': ''.join(from_local),
            'remote_self': self,
            'pid': os.getpid()


master <-----------------------------------------> remote
    --> send ---> |                   |  --> queue -->
                  | module:class/fqin |
    <-- queue <-- |                   |  <--- send <--

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

implant-0.1.2.tar.gz (34.3 kB view hashes)

Uploaded source

Built Distribution

implant-0.1.2-py3-none-any.whl (40.7 kB view hashes)

Uploaded py3

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Huawei Huawei PSF Sponsor Microsoft Microsoft PSF Sponsor NVIDIA NVIDIA PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page