通过Eureka获取微服务列表并检查heapdump泄露风险

通过以下python实现功能如下：

1. 通过解析http://2.1.2.13:8761/eureka/apps获取所有微服务信息，将http://{ip}:{port}格式字符串按行保存到targets.txt
2. 通过http://{ip}:{port}/heapdump判断所有微服务是否存在heapdump泄露风险，将存在该风险的url保存到
   vulnerable_heapdump.txt

import requests
import xml.etree.ElementTree as ET
from concurrent.futures import ThreadPoolExecutor, as_completed
import os

# ========================
# 配置参数
# ========================

EUREKA_URL = "http://2.1.2.13:8761/eureka/apps"  # 替换为你的 Eureka 地址
TARGETS_FILE = "targets.txt"
VULNERABLE_FILE = "vulnerable_heapdump.txt"
TIMEOUT = 5
THREADS = 30

HEADERS = {
    'Accept': 'application/xml',
    'User-Agent': 'HeapdumpScanner/1.0'
}

# ========================
# 从 Eureka 获取服务实例
# ========================

def fetch_eureka_services():
    print(f"[+] Fetching services from Eureka: {EUREKA_URL}")
    try:
        response = requests.get(EUREKA_URL, headers=HEADERS, timeout=TIMEOUT)
        response.raise_for_status()
        return response.text
    except Exception as e:
        print(f"[-] Failed to fetch from Eureka: {str(e)}")
        return None

def parse_eureka_xml(xml_data):
    root = ET.fromstring(xml_data)
    targets = []
    for app in root.findall('application'):
        print(app)
        for instance in app.findall('instance'):
            homePageUrl = instance.find('homePageUrl').text
            if homePageUrl is not None:
                targets.append(f"{homePageUrl}")

    return targets

def write_targets_to_file(targets):
    with open(TARGETS_FILE, 'w') as f:
        for target in targets:
            f.write(target + '\n')
    print(f"[+] Found {len(targets)} active instances. Saved to {TARGETS_FILE}")

# ========================
# 检测 /heapdump 接口
# ========================

def check_heapdump(url):
    full_url = f"{url}/heapdump"
    print(f"[+] Probing {full_url}")

    try:
        # 先用 HEAD 请求快速判断是否存在
        head_response = requests.head(full_url, timeout=TIMEOUT, allow_redirects=False)
        if head_response.status_code == 200:
            content_type = head_response.headers.get('Content-Type', '')
            if 'application/octet-stream' in content_type:
                print(f"[i] Potential heapdump (by Content-Type): {full_url}")
                return full_url

            # 如果不确定，再使用 Range 请求验证内容
            range_headers = {'Range': 'bytes=0-100'}
            get_response = requests.get(full_url, headers=range_headers, timeout=TIMEOUT, allow_redirects=False)

            if get_response.status_code in (200, 206):
                content = get_response.content
                if b'JAVA PROFILE' in content:
                    print(f"[!] Vulnerable heapdump endpoint found: {full_url}")
                    return full_url

        elif head_response.status_code == 401:
            print(f"[i] Auth required: {full_url}")
        elif head_response.status_code == 404:
            print(f"[-] Not found: {full_url}")
        else:
            print(f"[?] Unexpected status {head_response.status_code} on {full_url}")

    except Exception as e:
        print(f"[-] Error on {full_url}: {str(e)}")

    return None

def scan_targets():
    if not os.path.exists(TARGETS_FILE):
        print(f"[-] Targets file {TARGETS_FILE} not found.")
        return

    with open(TARGETS_FILE, 'r') as f:
        urls = [line.strip() for line in f if line.strip()]

    vulnerable_urls = []

    with ThreadPoolExecutor(max_workers=THREADS) as executor:
        future_to_url = {executor.submit(check_heapdump, url): url for url in urls}
        for future in as_completed(future_to_url):
            result = future.result()
            if result:
                vulnerable_urls.append(result)

    # 输出结果
    with open(VULNERABLE_FILE, 'w') as f:
        for url in vulnerable_urls:
            f.write(url + '\n')

    print(f"[+] Done. Found {len(vulnerable_urls)} vulnerable endpoints.")
    print(f"[+] Results saved to {VULNERABLE_FILE}")

# ========================
# 主程序入口
# ========================

def main():
    # 第一步：获取目标
    xml_data = fetch_eureka_services()
    # print(xml_data)
    if xml_data:
        targets = parse_eureka_xml(xml_data)
        write_targets_to_file(targets)

    # 第二步：扫描漏洞
    scan_targets()

if __name__ == '__main__':
    main()