DNS resolution analysis - IPv4 vs IPv6 investigation

gist_phase1_dns_resolution.py

#!/usr/bin/env python3
"""
Phase 1: Network-Level Deep Dive - DNS Resolution Testing
Tests DNS resolution behavior and IPv4/IPv6 connectivity
"""

import asyncio
import httpx
import socket
import ssl
import time
from typing import List, Dict, Any
import json

def get_dns_records(hostname: str) -> Dict[str, List[str]]:
    """Get all DNS records for a hostname"""
    records = {
        'ipv4': [],
        'ipv6': [],
        'all': []
    }
    
    try:
        # Get all address info
        addr_info = socket.getaddrinfo(hostname, 443, socket.AF_UNSPEC, socket.SOCK_STREAM)
        
        for family, _, _, _, sockaddr in addr_info:
            ip = sockaddr[0]
            records['all'].append(ip)
            
            if family == socket.AF_INET:
                if ip not in records['ipv4']:
                    records['ipv4'].append(ip)
            elif family == socket.AF_INET6:
                if ip not in records['ipv6']:
                    records['ipv6'].append(ip)
    except Exception as e:
        print(f"Error resolving {hostname}: {e}")
    
    return records

def test_direct_connection(ip: str, port: int, hostname: str, ipv6: bool = False) -> Dict[str, Any]:
    """Test direct connection to an IP address"""
    start_time = time.time()
    result = {
        'ip': ip,
        'ipv6': ipv6,
        'success': False,
        'error': None,
        'tls_version': None,
        'cipher': None
    }
    
    try:
        # Create socket
        sock = socket.socket(socket.AF_INET6 if ipv6 else socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(10)
        
        # Connect
        if ipv6:
            sock.connect((ip, port, 0, 0))
        else:
            sock.connect((ip, port))
        
        # Wrap with SSL
        context = ssl.create_default_context()
        with context.wrap_socket(sock, server_hostname=hostname) as ssock:
            result['success'] = True
            result['tls_version'] = ssock.version()
            result['cipher'] = ssock.cipher()
            
    except Exception as e:
        result['error'] = f"{type(e).__name__}: {str(e)}"
    
    result['duration'] = time.time() - start_time
    return result

def test_sync_with_ip(ip: str, ipv6: bool = False):
    """Test sync httpx client with specific IP"""
    print(f"\n🔄 Testing SYNC httpx.Client with {ip} ({'IPv6' if ipv6 else 'IPv4'})...")
    
    try:
        # Force specific IP by using custom transport
        transport = httpx.HTTPTransport(
            http2=False,
            retries=0,
            local_address=("::0" if ipv6 else "0.0.0.0", 0) if ipv6 else None
        )
        
        with httpx.Client(transport=transport, timeout=30.0) as client:
            # Connect directly to IP with Host header
            response = client.get(
                f"https://{ip}/public/video",
                params={"id": "7003402629929913605"},
                headers={
                    "Host": "api.tikapi.io",
                    "X-API-KEY": "YOUR_TIKAPI_KEY_HERE",
                    "Accept": "application/json"
                }
            )
            response.raise_for_status()
            print(f"  ✅ Success! Status: {response.status_code}")
            return True
    except Exception as e:
        print(f"  ❌ Failed: {type(e).__name__}: {str(e)}")
        return False

async def test_async_with_ip(ip: str, ipv6: bool = False):
    """Test async httpx client with specific IP"""
    print(f"\n⚡ Testing ASYNC httpx.AsyncClient with {ip} ({'IPv6' if ipv6 else 'IPv4'})...")
    
    try:
        # Force specific IP by using custom transport
        transport = httpx.AsyncHTTPTransport(
            http2=False,
            retries=0,
            local_address=("::0" if ipv6 else "0.0.0.0", 0) if ipv6 else None
        )
        
        async with httpx.AsyncClient(transport=transport, timeout=30.0) as client:
            # Connect directly to IP with Host header
            response = await client.get(
                f"https://{ip}/public/video",
                params={"id": "7003402629929913605"},
                headers={
                    "Host": "api.tikapi.io",
                    "X-API-KEY": "YOUR_TIKAPI_KEY_HERE",
                    "Accept": "application/json"
                }
            )
            response.raise_for_status()
            print(f"  ✅ Success! Status: {response.status_code}")
            return True
    except Exception as e:
        print(f"  ❌ Failed: {type(e).__name__}: {str(e)}")
        return False

def test_sync_with_family_preference():
    """Test sync httpx with different address family preferences"""
    print("\n🔄 Testing SYNC httpx.Client with family preferences...")
    
    results = []
    
    # Test with IPv4 only
    print("  Testing IPv4 only...")
    try:
        # Monkey-patch to force IPv4
        old_getaddrinfo = socket.getaddrinfo
        socket.getaddrinfo = lambda host, port, family=socket.AF_INET, *args, **kwargs: old_getaddrinfo(host, port, socket.AF_INET, *args, **kwargs)
        
        with httpx.Client(timeout=30.0, http2=False) as client:
            response = client.get(
                "https://api.tikapi.io/public/video",
                params={"id": "7003402629929913605"},
                headers={
                    "X-API-KEY": "YOUR_TIKAPI_KEY_HERE",
                    "Accept": "application/json"
                }
            )
            response.raise_for_status()
            results.append(("IPv4 only", True, None))
            print("    ✅ IPv4 only: Success")
    except Exception as e:
        results.append(("IPv4 only", False, str(e)))
        print(f"    ❌ IPv4 only: {e}")
    finally:
        socket.getaddrinfo = old_getaddrinfo
    
    # Test with IPv6 only
    print("  Testing IPv6 only...")
    try:
        # Monkey-patch to force IPv6
        old_getaddrinfo = socket.getaddrinfo
        socket.getaddrinfo = lambda host, port, family=socket.AF_INET6, *args, **kwargs: old_getaddrinfo(host, port, socket.AF_INET6, *args, **kwargs)
        
        with httpx.Client(timeout=30.0, http2=False) as client:
            response = client.get(
                "https://api.tikapi.io/public/video",
                params={"id": "7003402629929913605"},
                headers={
                    "X-API-KEY": "YOUR_TIKAPI_KEY_HERE",
                    "Accept": "application/json"
                }
            )
            response.raise_for_status()
            results.append(("IPv6 only", True, None))
            print("    ✅ IPv6 only: Success")
    except Exception as e:
        results.append(("IPv6 only", False, str(e)))
        print(f"    ❌ IPv6 only: {e}")
    finally:
        socket.getaddrinfo = old_getaddrinfo
    
    return results

async def test_async_with_family_preference():
    """Test async httpx with different address family preferences"""
    print("\n⚡ Testing ASYNC httpx.AsyncClient with family preferences...")
    
    results = []
    
    # Test with IPv4 only
    print("  Testing IPv4 only...")
    try:
        # Monkey-patch to force IPv4
        old_getaddrinfo = socket.getaddrinfo
        socket.getaddrinfo = lambda host, port, family=socket.AF_INET, *args, **kwargs: old_getaddrinfo(host, port, socket.AF_INET, *args, **kwargs)
        
        async with httpx.AsyncClient(timeout=30.0, http2=False) as client:
            response = await client.get(
                "https://api.tikapi.io/public/video",
                params={"id": "7003402629929913605"},
                headers={
                    "X-API-KEY": "YOUR_TIKAPI_KEY_HERE",
                    "Accept": "application/json"
                }
            )
            response.raise_for_status()
            results.append(("IPv4 only", True, None))
            print("    ✅ IPv4 only: Success")
    except Exception as e:
        results.append(("IPv4 only", False, str(e)))
        print(f"    ❌ IPv4 only: {e}")
    finally:
        socket.getaddrinfo = old_getaddrinfo
    
    # Test with IPv6 only
    print("  Testing IPv6 only...")
    try:
        # Monkey-patch to force IPv6
        old_getaddrinfo = socket.getaddrinfo
        socket.getaddrinfo = lambda host, port, family=socket.AF_INET6, *args, **kwargs: old_getaddrinfo(host, port, socket.AF_INET6, *args, **kwargs)
        
        async with httpx.AsyncClient(timeout=30.0, http2=False) as client:
            response = await client.get(
                "https://api.tikapi.io/public/video",
                params={"id": "7003402629929913605"},
                headers={
                    "X-API-KEY": "YOUR_TIKAPI_KEY_HERE",
                    "Accept": "application/json"
                }
            )
            response.raise_for_status()
            results.append(("IPv6 only", True, None))
            print("    ✅ IPv6 only: Success")
    except Exception as e:
        results.append(("IPv6 only", False, str(e)))
        print(f"    ❌ IPv6 only: {e}")
    finally:
        socket.getaddrinfo = old_getaddrinfo
    
    return results

async def main():
    """Run DNS resolution analysis"""
    print("🔍 Phase 1: Network-Level Deep Dive - DNS Resolution Testing")
    print("=" * 60)
    
    # Get DNS records
    print("\n📡 DNS Resolution for api.tikapi.io:")
    dns_records = get_dns_records("api.tikapi.io")
    print(f"  IPv4 addresses: {dns_records['ipv4']}")
    print(f"  IPv6 addresses: {dns_records['ipv6']}")
    
    # Test direct connections
    print("\n🔌 Testing Direct Socket Connections:")
    direct_results = []
    
    # Test IPv4 addresses
    for ip in dns_records['ipv4'][:2]:  # Test first 2 IPv4
        result = test_direct_connection(ip, 443, "api.tikapi.io", ipv6=False)
        print(f"  IPv4 {ip}: {'✅ Success' if result['success'] else f'❌ {result['error']}'}")
        if result['success']:
            print(f"    TLS: {result['tls_version']}, Cipher: {result['cipher'][0]}")
        direct_results.append(result)
    
    # Test IPv6 addresses
    for ip in dns_records['ipv6'][:2]:  # Test first 2 IPv6
        result = test_direct_connection(ip, 443, "api.tikapi.io", ipv6=True)
        print(f"  IPv6 {ip}: {'✅ Success' if result['success'] else f'❌ {result['error']}'}")
        if result['success']:
            print(f"    TLS: {result['tls_version']}, Cipher: {result['cipher'][0]}")
        direct_results.append(result)
    
    # Test httpx with family preferences
    print("\n🧪 Testing httpx with Address Family Preferences:")
    sync_family_results = test_sync_with_family_preference()
    async_family_results = await test_async_with_family_preference()
    
    # Test with specific IPs
    if dns_records['ipv4']:
        print("\n🎯 Testing httpx with Specific IPv4:")
        test_sync_with_ip(dns_records['ipv4'][0], ipv6=False)
        await test_async_with_ip(dns_records['ipv4'][0], ipv6=False)
    
    if dns_records['ipv6']:
        print("\n🎯 Testing httpx with Specific IPv6:")
        test_sync_with_ip(f"[{dns_records['ipv6'][0]}]", ipv6=True)
        await test_async_with_ip(f"[{dns_records['ipv6'][0]}]", ipv6=True)
    
    # Summary
    print("\n📊 Summary:")
    print("=" * 60)
    print("🔍 Key Findings:")
    
    # Check if IPv6 is the issue
    ipv4_works = any(r['success'] for r in direct_results if not r['ipv6'])
    ipv6_works = any(r['success'] for r in direct_results if r['ipv6'])
    
    if ipv4_works and not ipv6_works:
        print("  ⚠️  IPv6 connectivity appears to be the issue!")
        print("  ✅ IPv4 connections work fine")
        print("  ❌ IPv6 connections fail")
    elif ipv4_works and ipv6_works:
        print("  ✅ Both IPv4 and IPv6 direct connections work")
        print("  🤔 Issue may be with httpx's IPv6 handling")
    else:
        print("  ❓ Unexpected results - further investigation needed")
    
    # Save results
    with open('dns_resolution_results.json', 'w') as f:
        json.dump({
            'dns_records': dns_records,
            'direct_connection_results': direct_results,
            'sync_family_results': sync_family_results,
            'async_family_results': async_family_results
        }, f, indent=2, default=str)
    
    print("\n💾 Detailed results saved to dns_resolution_results.json")

if __name__ == "__main__":
    asyncio.run(main())