wtf-ip.py

#!/usr/bin/env python3
"""
╦ ╦╔╦╗╔═╗   ╦╔═╗
║║║ ║ ╠╣ ╔╦╝║╠═╝
╚╩╝ ╩ ╚   ╩ ╩╩  

WTF-IP: Who's That From - IP Address Analyzer
==============================================

A command-line tool to determine if IP addresses are likely bots/datacenters 
or human/residential users by analyzing WHOIS data, ASN information, and 
network ownership patterns.

Author: Generated for edwilde
Date: 2025-11-26
License: MIT

Usage:
    ./wtf-ip.py
    
    Then paste IP addresses (one per line) with optional counts
    Press Ctrl+D (Unix) or Ctrl+Z (Windows) when done

Example Input:
    169.47.39.105	148     (with count)
    188.241.60.103	123     (with count)
    104.153.67.10           (no count - defaults to 1)
    206.72.194.37           (no count - defaults to 1)

Output:
    Detailed report showing organization, country, ASN, classification 
    (LIKELY BOT vs LIKELY HUMAN), and summary statistics.
"""

import sys
import re
import json
import subprocess
from typing import Dict, Tuple, List
from collections import OrderedDict

# =============================================================================
# CONFIGURATION: Detection Patterns
# =============================================================================

# Known cloud/hosting/datacenter ASN patterns and providers
# These are strong indicators of bot/automated traffic from datacenters
KNOWN_CLOUD_PROVIDERS = [
    'amazon', 'aws', 'azure', 'microsoft', 'google', 'gcp', 'digitalocean',
    'linode', 'vultr', 'ovh', 'hetzner', 'cloudflare', 'akamai', 'fastly',
    'rackspace', 'softlayer', 'ibm', 'oracle', 'alibaba', 'tencent',
    'datacentr', 'datacenter', 'data center', 'hosting', 'server', 'cloud',
    'vps', 'virtual', 'dedicated', 'colocation', 'colo', 'vpn', 'proxy',
    'contabo', 'serverspace', 'serverius', 'selectel'
]

# Known residential/consumer ISP indicators
# These suggest legitimate human traffic from home/mobile connections
KNOWN_RESIDENTIAL_INDICATORS = [
    'broadband', 'cable', 'dsl', 'fiber', 'mobile', 'wireless', 
    'telecom', 'communications', 'residential', 'consumer', 'internet service',
    'isp', 'comcast', 'verizon', 'att', 'at&t', 'spectrum', 'cox',
    'bt group', 'virgin media', 'sky broadband', 'vodafone', 'orange',
    'telefonica', 'telekom', 'telstra', 'bell canada', 'rogers'
]

# Country code to flag emoji mapping
def country_flag(country_code: str) -> str:
    """
    Convert 2-letter country code to flag emoji.
    
    Args:
        country_code: Two-letter ISO country code (e.g., 'US', 'GB')
        
    Returns:
        Flag emoji for the country, or empty string if invalid
    """
    if not country_code or len(country_code) != 2:
        return ""
    
    # Convert country code to flag emoji using regional indicator symbols
    # A=🇦 (U+1F1E6), B=🇧 (U+1F1E7), etc.
    country_code = country_code.upper()
    return ''.join(chr(0x1F1E6 + ord(char) - ord('A')) for char in country_code)


# =============================================================================
# WHOIS LOOKUP FUNCTIONS
# =============================================================================

def run_whois(ip: str) -> str:
    """
    Run whois command and return output.
    
    Args:
        ip: IP address to look up
        
    Returns:
        WHOIS output as string, or error message
    """
    try:
        result = subprocess.run(['whois', ip], capture_output=True, text=True, timeout=10)
        return result.stdout
    except Exception as e:
        return f"Error: {str(e)}"


def parse_whois_info(whois_output: str) -> Dict[str, str]:
    """
    Extract key information from WHOIS output.
    
    Parses WHOIS data to extract organization name, network name, country,
    ASN number, and ASN description. Handles multiple WHOIS formats.
    
    Args:
        whois_output: Raw WHOIS command output
        
    Returns:
        Dictionary containing parsed WHOIS information
    """
    info = {
        'org': '',
        'netname': '',
        'country': '',
        'asn': '',
        'asn_name': '',
        'org_type': ''
    }
    
    lines = whois_output.split('\n')
    asn_descriptions = []  # Collect all descriptions for better analysis
    
    for line in lines:
        line_lower = line.lower()
        
        # Organization
        if re.match(r'^(org-name|orgname|organization|org):\s*', line_lower):
            if not info['org']:  # Take first match
                info['org'] = re.sub(r'^[^:]+:\s*', '', line).strip()
        
        # Network name
        elif re.match(r'^(netname|network-name):\s*', line_lower):
            info['netname'] = re.sub(r'^[^:]+:\s*', '', line).strip()
        
        # Country
        elif re.match(r'^(country|geoloc):\s*', line_lower):
            if not info['country']:  # Take first match
                info['country'] = re.sub(r'^[^:]+:\s*', '', line).strip()[:2].upper()
        
        # ASN
        elif re.match(r'^(origin|originas|as-number|asn):\s*', line_lower):
            asn_match = re.search(r'AS(\d+)', line, re.IGNORECASE)
            if asn_match and not info['asn']:
                info['asn'] = asn_match.group(1)
        
        # ASN Name/Description - collect all for better classification
        elif re.match(r'^(as-name|asname|descr|description):\s*', line_lower):
            desc = re.sub(r'^[^:]+:\s*', '', line).strip()
            if desc and desc not in asn_descriptions:
                asn_descriptions.append(desc)
                if not info['asn_name']:
                    info['asn_name'] = desc
    
    # Try to extract ASN from anywhere in the output if not found
    if not info['asn']:
        asn_match = re.search(r'\b(AS|ASN)[\s:-]*(\d+)\b', whois_output, re.IGNORECASE)
        if asn_match:
            info['asn'] = asn_match.group(2)
    
    # Fallback: use org as ASN name if not found
    if not info['asn_name'] and info['org']:
        info['asn_name'] = info['org']
    
    # Store all descriptions for better classification
    info['all_descriptions'] = ' '.join(asn_descriptions) if asn_descriptions else ''
    
    # Determine org type
    info['org_type'] = classify_org_type(
        info['org'], 
        info['netname'], 
        info['asn_name'],
        info.get('all_descriptions', '')
    )
    
    return info


def classify_org_type(org: str, netname: str, asn_name: str, all_desc: str = '') -> str:
    """
    Classify organization as ISP, Cloud/Hosting, or Unknown.
    
    Checks organization names against known patterns to determine if the
    IP is from a cloud provider/datacenter or residential ISP.
    
    Args:
        org: Organization name from WHOIS
        netname: Network name from WHOIS
        asn_name: ASN description from WHOIS
        all_desc: All descriptions combined for better matching
        
    Returns:
        Classification string: CLOUD/HOSTING, ISP/RESIDENTIAL, or UNKNOWN
    """
    combined = f"{org} {netname} {asn_name} {all_desc}".lower()
    
    # Check for cloud/hosting indicators first (more specific)
    for indicator in KNOWN_CLOUD_PROVIDERS:
        if indicator in combined:
            return "CLOUD/HOSTING"
    
    # Check for residential/ISP indicators
    for indicator in KNOWN_RESIDENTIAL_INDICATORS:
        if indicator in combined:
            return "ISP/RESIDENTIAL"
    
    # Additional heuristics for unknowns
    # If it has "network" or "net" without other indicators, likely ISP
    if re.search(r'\b(networks?|netcom|netwerk)\b', combined) and 'host' not in combined:
        return "ISP/RESIDENTIAL"
    
    # If it mentions "access" or "broadband", likely residential
    if 'access' in combined or 'broadband' in combined:
        return "ISP/RESIDENTIAL"
    
    # Default to unknown
    return "UNKNOWN"


def determine_likelihood(org_type: str) -> str:
    """
    Determine if traffic is likely bot or human based on org type.
    
    Args:
        org_type: Organization classification (CLOUD/HOSTING, ISP/RESIDENTIAL, or UNKNOWN)
        
    Returns:
        Human-readable likelihood string with emoji
    """
    if org_type == "CLOUD/HOSTING":
        return "🤖 LIKELY BOT"
    elif org_type == "ISP/RESIDENTIAL":
        return "👤 LIKELY HUMAN"
    else:
        return "❓ UNKNOWN"


def analyze_ip(ip: str, count: int, show_progress: bool = False, current: int = 0, total: int = 0) -> Dict:
    """
    Analyze a single IP address.
    
    Performs WHOIS lookup, parses results, and classifies the IP.
    
    Args:
        ip: IP address to analyze
        count: Request count for this IP
        show_progress: Whether to show progress indicator
        current: Current IP number (for progress)
        total: Total number of IPs (for progress)
        
    Returns:
        Dictionary containing all analysis results
    """
    if show_progress and total > 0:
        # Show progress bar for multiple IPs
        progress = int((current / total) * 30)  # 30 char progress bar
        bar = '█' * progress + '░' * (30 - progress)
        percent = int((current / total) * 100)
        print(f"  [{current}/{total}] {bar} {percent}% | {ip}...", end='\r', file=sys.stderr)
    else:
        print(f"  Analyzing {ip}...", end='\r', file=sys.stderr)
    
    whois_output = run_whois(ip)
    info = parse_whois_info(whois_output)
    info['ip'] = ip
    info['count'] = count
    info['likelihood'] = determine_likelihood(info['org_type'])
    
    return info



# =============================================================================
# REPORTING FUNCTIONS
# =============================================================================

def format_report(results: List[Dict]) -> str:
    """
    Format analysis results into a readable report.
    
    Creates a detailed report showing each IP's classification and summary stats.
    
    Args:
        results: List of analysis result dictionaries
        
    Returns:
        Formatted report string
    """
    report = []
    report.append("=" * 100)
    report.append("IP ADDRESS ANALYSIS REPORT")
    report.append("=" * 100)
    report.append("")
    
    for idx, result in enumerate(results, 1):
        # Format country with flag emoji
        country_display = result['country'] or 'N/A'
        if result['country']:
            flag = country_flag(result['country'])
            country_display = f"{flag} {result['country']}" if flag else result['country']
        
        report.append(f"{idx}. {result['ip']} (Count: {result['count']})")
        report.append(f"   Classification: {result['likelihood']}")
        report.append(f"   Organization:   {result['org'] or result['netname'] or 'N/A'}")
        report.append(f"   Country:        {country_display}")
        report.append(f"   ASN:            AS{result['asn']} - {result['asn_name']}" if result['asn'] else "   ASN:            N/A")
        report.append(f"   Type:           {result['org_type']}")
        report.append("")
    
    # Summary statistics
    report.append("=" * 100)
    report.append("SUMMARY")
    report.append("=" * 100)
    
    total_ips = len(results)
    bots = sum(1 for r in results if "BOT" in r['likelihood'])
    humans = sum(1 for r in results if "HUMAN" in r['likelihood'])
    unknown = sum(1 for r in results if "UNKNOWN" in r['likelihood'])
    
    total_count = sum(r['count'] for r in results)
    bot_count = sum(r['count'] for r in results if "BOT" in r['likelihood'])
    human_count = sum(r['count'] for r in results if "HUMAN" in r['likelihood'])
    unknown_count = sum(r['count'] for r in results if "UNKNOWN" in r['likelihood'])
    
    report.append(f"Total IPs:        {total_ips}")
    report.append(f"  Likely Bots:    {bots} ({bots/total_ips*100:.1f}%)")
    report.append(f"  Likely Humans:  {humans} ({humans/total_ips*100:.1f}%)")
    report.append(f"  Unknown:        {unknown} ({unknown/total_ips*100:.1f}%)")
    report.append("")
    report.append(f"Total Requests:   {total_count}")
    report.append(f"  Bot Requests:   {bot_count} ({bot_count/total_count*100:.1f}%)")
    report.append(f"  Human Requests: {human_count} ({human_count/total_count*100:.1f}%)")
    report.append(f"  Unknown:        {unknown_count} ({unknown_count/total_count*100:.1f}%)")
    report.append("=" * 100)
    
    return "\n".join(report)



# =============================================================================
# INPUT PARSING
# =============================================================================

def parse_input(input_text: str) -> List[Tuple[str, int]]:
    """
    Parse IP addresses and counts from input text.
    
    Supports two formats:
      - IP<whitespace>count (e.g., "1.1.1.1  50")
      - IP only (e.g., "1.1.1.1") - count defaults to 1
    
    Args:
        input_text: Raw input text containing IPs with optional counts
        
    Returns:
        List of (ip, count) tuples
    """
    ips = []
    for line in input_text.strip().split('\n'):
        line = line.strip()
        if not line:
            continue
        
        # Try to match IP with count first
        match_with_count = re.match(r'^([\d.]+)\s+(\d+)$', line)
        if match_with_count:
            ip = match_with_count.group(1)
            count = int(match_with_count.group(2))
            ips.append((ip, count))
            continue
        
        # Try to match IP only (no count)
        match_ip_only = re.match(r'^([\d.]+)$', line)
        if match_ip_only:
            ip = match_ip_only.group(1)
            ips.append((ip, 1))  # Default count to 1
            continue
    
    return ips



# =============================================================================
# MAIN PROGRAM
# =============================================================================

def main():
    """Main program entry point."""
    # Display masthead
    print("╦ ╦╔╦╗╔═╗   ╦╔═╗", file=sys.stderr)
    print("║║║ ║ ╠╣ ╔╦╝║╠═╝", file=sys.stderr)
    print("╚╩╝ ╩ ╚   ╩ ╩╩  ", file=sys.stderr)
    print("", file=sys.stderr)
    print("IP Address Bot/Human Analyzer", file=sys.stderr)
    print("=" * 50, file=sys.stderr)
    print("Paste IP addresses (one per line) or with counts:", file=sys.stderr)
    print("  Format: IP or IP<tab>count", file=sys.stderr)
    print("  Examples: 1.1.1.1 or 1.1.1.1  50", file=sys.stderr)
    print("Press Ctrl+D (Unix) or Ctrl+Z (Windows) when done:", file=sys.stderr)
    print("", file=sys.stderr)
    
    # Read input from stdin
    try:
        input_text = sys.stdin.read()
    except KeyboardInterrupt:
        print("\nCancelled.", file=sys.stderr)
        sys.exit(1)
    
    # Parse IPs
    ip_list = parse_input(input_text)
    
    if not ip_list:
        print("No valid IP addresses found in input.", file=sys.stderr)
        sys.exit(1)
    
    print(f"\nFound {len(ip_list)} IP address{'' if len(ip_list) == 1 else 'es'} to analyze...\n", file=sys.stderr)
    
    # Determine if we should show progress bar (>3 IPs)
    show_progress = len(ip_list) > 3
    
    # Analyze each IP
    results = []
    for idx, (ip, count) in enumerate(ip_list, 1):
        result = analyze_ip(ip, count, show_progress=show_progress, current=idx, total=len(ip_list))
        results.append(result)
    
    # Clear progress line with enough spaces
    print(" " * 80, file=sys.stderr, end='\r')
    print("\nAnalysis complete!\n", file=sys.stderr)
    
    # Print report
    report = format_report(results)
    print(report)
    
    # Offer to copy to clipboard
    print("\n", file=sys.stderr)
    print("Press 'c' to copy report to clipboard, or any other key to exit: ", file=sys.stderr, end='', flush=True)
    
    try:
        # Try to read from terminal directly
        import termios
        import tty
        
        # Open terminal directly for input
        with open('/dev/tty', 'r') as tty_in:
            fd = tty_in.fileno()
            old_settings = termios.tcgetattr(fd)
            
            try:
                # Set terminal to raw mode to read single character
                tty.setraw(fd)
                char = tty_in.read(1).lower()
            finally:
                # Restore terminal settings
                termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
        
        print("", file=sys.stderr)  # New line after input
        
        if char == 'c':
            # Copy to clipboard using pbcopy
            try:
                process = subprocess.run(
                    ['pbcopy'],
                    input=report,
                    text=True,
                    capture_output=True,
                    check=True
                )
                print("✅ Report copied to clipboard!", file=sys.stderr)
            except subprocess.CalledProcessError:
                print("❌ Error: Failed to copy to clipboard.", file=sys.stderr)
            except FileNotFoundError:
                print("❌ Error: pbcopy not found. Clipboard copy only works on macOS.", file=sys.stderr)
        else:
            print("Exiting without copying.", file=sys.stderr)
    
    except Exception as e:
        # Fallback if terminal manipulation fails
        print(f"\n(Clipboard copy feature unavailable: {e})", file=sys.stderr)


if __name__ == '__main__':
    main()