diff --git a/iplookup.py b/iplookup.py index 55333cf..4037275 100644 --- a/iplookup.py +++ b/iplookup.py @@ -4,16 +4,19 @@ Features: Multi-source aggregation, security analysis, performance testing, export capabilities """ +from __future__ import annotations import requests import json import sys import time import socket import argparse +import os +import subprocess +import re from datetime import datetime -from typing import Dict, List, Optional, Tuple +from typing import Dict, List, Optional, Tuple, Any import concurrent.futures -from urllib.parse import urlparse # Try to import optional dependencies try: @@ -44,29 +47,54 @@ def __init__(self, verbose=False, timeout=10): 'User-Agent': 'Ultimate-IP-Lookup-Tool/2.0' }) + # --- HACKER LOOK OVERHAUL --- def print_banner(self): - """Display awesome ASCII banner""" - banner = """ -╔═══════════════════════════════════════════════════════════════╗ -║ ║ -║ ██╗██████╗ ██╗ ██████╗ ██████╗ ██╗ ██╗██╗ ██╗██████╗ ║ -║ ██║██╔══██╗ ██║ ██╔═══██╗██╔═══██╗██║ ██╔╝██║ ██║██╔══██╗ ║ -║ ██║██████╔╝ ██║ ██║ ██║██║ ██║█████╔╝ ██║ ██║██████╔╝ ║ -║ ██║██╔═══╝ ██║ ██║ ██║██║ ██║██╔═██╗ ██║ ██║██╔═══╝ ║ -║ ██║██║ ███████╗╚██████╔╝╚██████╔╝██║ ██╗╚██████╔╝██║ ║ -║ ╚═╝╚═╝ ╚══════╝ ╚═════╝ ╚═════╝ ╚═╝ ╚═╝ ╚═════╝ ╚═╝ ║ -║ ║ -║ Ultimate IP Intelligence & Analysis ║ -║ v2.0.0 - 2025 ║ -╚═══════════════════════════════════════════════════════════════╝ - """ + """Display stylized NEON ASCII banner.""" + + # New, high-contrast ASCII art for a "hacker" feel + banner = r""" + ╔═══════════════════════════════════════════════════════════════╗ + ║ ║ + ║ ██╗██████╗ ██╗ ██████╗ ██████╗ ██╗ ██╗██╗ ██╗██████╗ ║ + ║ ██║██╔══██╗ ██║ ██╔═══██╗██╔═══██╗██║ ██╔╝██║ ██║██╔══██╗ ║ + ║ ██║██████╔╝ ██║ ██║ ██║██║ ██║█████╔╝ ██║ ██║██████╔╝ ║ + ║ ██║██╔═══╝ ██║ ██║ ██║██║ ██║██╔═██╗ ██║ ██║██╔═══╝ ║ + ║ ██║██║ ███████╗╚██████╔╝╚██████╔╝██║ ██╗╚██████╔╝██║ ║ + ║ ╚═╝╚═╝ ╚══════╝ ╚═════╝ ╚═════╝ ╚═╝ ╚═╝ ╚═════╝ ╚═╝ ║ + ║ ║ + ║ Ultimate IP Intelligence & Analysis ║ + ║ v2.0.0 - 2025 ║ + ╚═══════════════════════════════════════════════════════════════╝ +""" + # Using specific color styles: bright_cyan for the frame, bold_green for text. if RICH_AVAILABLE: - console.print(banner, style="bold cyan") + console.print( + Panel( + banner, + title="[bold magenta]STATUS: ACTIVE[/bold magenta]", + border_style="bright_cyan", + box=box.DOUBLE_EDGE + ), + style="bold green" + ) else: print(banner) + def _get_ip_from_source(self, source: str) -> Optional[str]: + """Worker function to fetch IP from a single source.""" + try: + response = self.session.get(source, timeout=3) # Shorter timeout for quick sources + response.raise_for_status() + + if 'json' in source: + return response.json().get('ip') + else: + return response.text.strip() + except Exception: + return None + def get_public_ip(self) -> Optional[str]: - """Get public IP with fallback sources""" + """Get public IP with concurrent fallback sources (Parallelized)""" sources = [ 'https://api.ipify.org?format=json', 'https://api64.ipify.org?format=json', @@ -75,25 +103,23 @@ def get_public_ip(self) -> Optional[str]: 'https://ident.me' ] - for source in sources: - try: - response = self.session.get(source, timeout=self.timeout) - response.raise_for_status() - - if 'json' in source: - return response.json().get('ip') - else: - return response.text.strip() - except Exception as e: - if self.verbose: - print(f"Failed to get IP from {source}: {e}") - continue + # Use ThreadPoolExecutor to run all requests concurrently + with concurrent.futures.ThreadPoolExecutor(max_workers=len(sources)) as executor: + futures = [executor.submit(self._get_ip_from_source, source) for source in sources] + + for future in concurrent.futures.as_completed(futures): + ip = future.result() + if ip: + # Cancel any remaining futures and return the first successful result + executor.shutdown(wait=False, cancel_futures=True) + return ip return None def get_ip_api_data(self, ip: str) -> Dict: - """Get data from ip-api.com""" - url = f'http://ip-api.com/json/{ip}?fields=66846719' + """Get data from ip-api.com (Uses human-readable fields)""" + fields = "status,message,continent,country,countryCode,region,regionName,city,zip,lat,lon,timezone,isp,org,as,query,mobile,proxy,hosting" + url = f'http://ip-api.com/json/{ip}?fields={fields}' try: response = self.session.get(url, timeout=self.timeout) response.raise_for_status() @@ -135,7 +161,12 @@ def get_abuse_ipdb_data(self, ip: str, api_key: Optional[str] = None) -> Dict: response.raise_for_status() return response.json().get('data', {}) except Exception as e: - return {'error': str(e)} + # Check for specific API error messages + try: + err_json = response.json() + return {'error': err_json.get('errors')[0].get('detail', 'API Error')} + except: + return {'error': str(e)} def reverse_dns_lookup(self, ip: str) -> str: """Perform reverse DNS lookup""" @@ -148,28 +179,52 @@ def reverse_dns_lookup(self, ip: str) -> str: return f"Error: {str(e)}" def ping_test(self, ip: str, count: int = 4) -> Dict: - """Perform ping test (HTTP-based)""" - results = [] - for _ in range(count): - try: - start = time.time() - response = self.session.get(f'http://{ip}', timeout=2) - latency = (time.time() - start) * 1000 - results.append(latency) - except: - results.append(None) + """Perform system ICMP ping test (True ICMP test)""" + param = '-n' if os.name == 'nt' else '-c' + command = ['ping', param, str(count), ip] - valid_results = [r for r in results if r is not None] - if valid_results: - return { - 'sent': count, - 'received': len(valid_results), - 'lost': count - len(valid_results), - 'min': min(valid_results), - 'max': max(valid_results), - 'avg': sum(valid_results) / len(valid_results) - } - return {'error': 'All pings failed'} + try: + # Run ping command + result = subprocess.run( + command, + capture_output=True, + text=True, + timeout=self.timeout, + check=True # Raise error for non-zero exit code (e.g., target unreachable) + ) + + # Parse output (simplified for reliability across OS) + output = result.stdout + + # Find RTT min/avg/max statistics + match = re.search(r'min/avg/max/(stddev)? = ([\d.]+)/([\d.]+)/([\d.]+)', output) + + if match: + stats = match.groups() + return { + 'sent': count, + 'received': count, # Assuming successful run implies all sent/received in basic test + 'lost': 0, + 'min': float(stats[1]), + 'avg': float(stats[2]), + 'max': float(stats[3]) + } + + # Fallback for systems where regex fails (e.g., Mac OS X, non-English) + if '100% packet loss' in output: + return {'error': '100% packet loss'} + + return {'error': 'Failed to parse ping output'} + + except subprocess.TimeoutExpired: + return {'error': f'Ping timeout after {self.timeout}s'} + except subprocess.CalledProcessError as e: + # Target unreachable or network error + if '100% packet loss' in e.stdout: + return {'error': '100% packet loss'} + return {'error': 'Target unreachable'} + except Exception as e: + return {'error': f'System error: {str(e)}'} def check_ports(self, ip: str, ports: List[int] = None) -> Dict[int, bool]: """Check common ports""" @@ -181,6 +236,7 @@ def check_ports(self, ip: str, ports: List[int] = None) -> Dict[int, bool]: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(2) try: + # Use connect_ex for non-blocking result result = sock.connect_ex((ip, port)) results[port] = result == 0 except: @@ -190,19 +246,23 @@ def check_ports(self, ip: str, ports: List[int] = None) -> Dict[int, bool]: return results - def aggregate_data(self, ip: str, abuse_api_key: Optional[str] = None) -> Dict: - """Aggregate data from multiple sources""" - aggregated = {'ip': ip, 'timestamp': datetime.now().isoformat()} + def aggregate_data(self, ip: str, abuse_api_key: Optional[str] = None, show_ports: bool = False, show_ping: bool = False) -> Dict: + """Aggregate data from multiple sources (Parallelized)""" + aggregated: Dict[str, Any] = {'ip': ip, 'timestamp': datetime.now().isoformat()} + max_workers = 5 + if show_ports: max_workers += 1 + if show_ping: max_workers += 1 + if RICH_AVAILABLE: with Progress( - SpinnerColumn(), + SpinnerColumn(spinner_name="dots"), TextColumn("[progress.description]{task.description}"), console=console ) as progress: - task = progress.add_task("[cyan]Gathering intelligence...", total=None) + progress.add_task("[bold yellow]Gathering intelligence...[/bold yellow]", total=None) - with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: + with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: futures = { 'ip_api': executor.submit(self.get_ip_api_data, ip), 'ipinfo': executor.submit(self.get_ipinfo_data, ip), @@ -212,6 +272,12 @@ def aggregate_data(self, ip: str, abuse_api_key: Optional[str] = None) -> Dict: if abuse_api_key: futures['abuse_ipdb'] = executor.submit(self.get_abuse_ipdb_data, ip, abuse_api_key) + + if show_ports: + futures['ports'] = executor.submit(self.check_ports, ip) + + if show_ping: + futures['ping'] = executor.submit(self.ping_test, ip) for key, future in futures.items(): aggregated[key] = future.result() @@ -224,129 +290,136 @@ def aggregate_data(self, ip: str, abuse_api_key: Optional[str] = None) -> Dict: if abuse_api_key: aggregated['abuse_ipdb'] = self.get_abuse_ipdb_data(ip, abuse_api_key) - + + if show_ports: + aggregated['ports'] = self.check_ports(ip) + + if show_ping: + aggregated['ping'] = self.ping_test(ip) + return aggregated - def display_results(self, data: Dict, show_ports: bool = False, show_ping: bool = False): - """Display results with rich formatting""" + def display_results(self, data: Dict): + """Display results with rich formatting (Enhanced Hacker Look)""" if not RICH_AVAILABLE: - self._display_results_plain(data, show_ports, show_ping) + self._display_results_plain(data) return ip = data.get('ip') + # --- Layout setup for clean separation --- + layout = Layout(name="root") + layout.split_column( + Layout(name="main_info", size=20), + Layout(name="bottom") + ) + layout["bottom"].split_row( + Layout(name="security", size=45), + Layout(name="diagnostics") + ) + # Main Info Panel ip_api = data.get('ip_api', {}) + ipinfo = data.get('ipinfo', {}) + if ip_api.get('status') == 'success': info_text = f""" -[bold cyan]IP Address:[/bold cyan] {ip} -[bold cyan]Hostname:[/bold cyan] {data.get('reverse_dns', 'N/A')} -[bold cyan]Type:[/bold cyan] {ip_api.get('query', 'N/A')} +[bold bright_white]TARGET IP:[/bold bright_white] [yellow]{ip}[/yellow] +[bold bright_white]Reverse DNS:[/bold bright_white] [dim]{data.get('reverse_dns', 'N/A')}[/dim] +[bold bright_white]Geolocation Type:[/bold bright_white] [dim]{ip_api.get('query', 'N/A')} ({ip_api.get('continent', 'N/A')})[/dim] -[bold yellow]📍 Location[/bold yellow] - Country: {ip_api.get('country', 'N/A')} ({ip_api.get('countryCode', 'N/A')}) - Region: {ip_api.get('regionName', 'N/A')} - City: {ip_api.get('city', 'N/A')} - ZIP: {ip_api.get('zip', 'N/A')} - Timezone: {ip_api.get('timezone', 'N/A')} - Coordinates: {ip_api.get('lat', 'N/A')}, {ip_api.get('lon', 'N/A')} +[bold #00ff00]>> LOCATION DATA[/bold #00ff00] + [dim]Country:[/dim] {ip_api.get('country', 'N/A')} ([dim]{ip_api.get('countryCode', 'N/A')}[/dim]) + [dim]City/Region:[/dim] {ip_api.get('city', 'N/A')}, {ip_api.get('regionName', 'N/A')} + [dim]Timezone:[/dim] [dim]{ip_api.get('timezone', 'N/A')}[/dim] + [dim]Coordinates:[/dim] [dim]{ip_api.get('lat', 'N/A')}, {ip_api.get('lon', 'N/A')}[/dim] -[bold green]🌐 Network[/bold green] - ISP: {ip_api.get('isp', 'N/A')} - Organization: {ip_api.get('org', 'N/A')} - AS: {ip_api.get('as', 'N/A')} - Mobile: {ip_api.get('mobile', False)} - Proxy: {ip_api.get('proxy', False)} - Hosting: {ip_api.get('hosting', False)} +[bold #00ffff]>> INFRASTRUCTURE[/bold #00ffff] + [dim]ISP:[/dim] {ip_api.get('isp', 'N/A')} + [dim]Organization:[/dim] {ip_api.get('org', 'N/A')} + [dim]AS Number:[/dim] {ip_api.get('as', 'N/A')} + [dim]Tags (Mobile/Proxy/Host):[/dim] [yellow]{ip_api.get('mobile', False)}[/yellow] / [yellow]{ip_api.get('proxy', False)}[/yellow] / [yellow]{ip_api.get('hosting', False)}[/yellow] """ - console.print(Panel(info_text, title="[bold]IP Intelligence Report[/bold]", border_style="cyan")) + layout["main_info"].update(Panel(info_text, title="[bold bright_white]-- IP Intelligence Core --[/bold bright_white]", border_style="bright_cyan", box=box.HEAVY)) # Security Analysis abuse_data = data.get('abuse_ipdb', {}) if abuse_data and 'error' not in abuse_data: abuse_score = abuse_data.get('abuseConfidenceScore', 0) - is_whitelisted = abuse_data.get('isWhitelisted', False) - total_reports = abuse_data.get('totalReports', 0) - if abuse_score > 75: - risk_level = "[bold red]HIGH RISK ⚠️[/bold red]" + if abuse_score > 85: + risk_level = "[bold white on red]HIGH RISK - BLOCKED 🚨[/bold white on red]" style = "red" - elif abuse_score > 25: + elif abuse_score > 40: risk_level = "[bold yellow]MODERATE RISK ⚡[/bold yellow]" style = "yellow" else: - risk_level = "[bold green]LOW RISK ✓[/bold green]" - style = "green" + risk_level = "[bold #00ff00]LOW RISK ✓[/bold #00ff00]" + style = "#00ff00" security_text = f""" -{risk_level} +{risk_level} -Abuse Confidence Score: {abuse_score}% -Total Reports: {total_reports} -Whitelisted: {is_whitelisted} -Last Reported: {abuse_data.get('lastReportedAt', 'Never')} +[dim]Abuse Score:[/dim] [bold]{abuse_score}%[/bold] +[dim]Total Reports:[/dim] {abuse_data.get('totalReports', 0)} +[dim]Whitelisted:[/dim] {abuse_data.get('isWhitelisted', False)} +[dim]Last Reported:[/dim] [dim]{abuse_data.get('lastReportedAt', 'Never')}[/dim] """ - console.print(Panel(security_text, title="[bold]🔒 Security Analysis[/bold]", border_style=style)) + layout["security"].update(Panel(security_text, title="[bold red]-- Threat Assessment --[/bold red]", border_style=style, box=box.ROUNDED)) + elif 'error' in abuse_data: + layout["security"].update(Panel(f"[dim]Security data not available:[/dim]\n[red]{abuse_data['error']}[/red]", title="[bold red]-- Threat Assessment --[/bold red]", border_style="dim", box=box.ROUNDED)) - # Additional Data Sources - ipinfo = data.get('ipinfo', {}) - if ipinfo and 'error' not in ipinfo: - table = Table(title="Additional Information", box=box.ROUNDED) - table.add_column("Source", style="cyan") - table.add_column("Data", style="white") - - if ipinfo.get('hostname'): - table.add_row("IPInfo Hostname", ipinfo.get('hostname')) - if ipinfo.get('org'): - table.add_row("IPInfo Org", ipinfo.get('org')) - if ipinfo.get('postal'): - table.add_row("Postal Code", ipinfo.get('postal')) - - console.print(table) + # Diagnostics (Ports, Ping, and WHOIS) + # --- FIX: Ensure content is always present --- + diag_tree = Tree("[bold bright_magenta]>> System Diagnostics[/bold bright_magenta]", guide_style="dim") - # Port Scan Results - if show_ports: - ip_str = data.get('ip') - if ip_str: - console.print("\n[bold cyan]Scanning common ports...[/bold cyan]") - ports = self.check_ports(ip_str) - - port_table = Table(title="🔍 Port Scan Results", box=box.ROUNDED) - port_table.add_column("Port", style="cyan") - port_table.add_column("Service", style="yellow") - port_table.add_column("Status", style="white") - - port_services = { - 21: "FTP", 22: "SSH", 23: "Telnet", 25: "SMTP", - 80: "HTTP", 443: "HTTPS", 3306: "MySQL", 3389: "RDP", - 5432: "PostgreSQL", 8080: "HTTP-Alt" - } - - for port, is_open in ports.items(): - service = port_services.get(port, "Unknown") - status = "[green]OPEN[/green]" if is_open else "[red]CLOSED[/red]" - port_table.add_row(str(port), service, status) - - console.print(port_table) + # 0. Always present: Time & Status + now = datetime.fromisoformat(data['timestamp']).strftime("%H:%M:%S") + diag_tree.add(f"[bold dim]STATUS:[/bold dim] [bold #00ff00]SCAN COMPLETE[/bold #00ff00] ([dim]{now}[/dim])") - # Ping Test - if show_ping: - ip_str = data.get('ip') - if ip_str: - console.print("\n[bold cyan]Performing connectivity test...[/bold cyan]") - ping_results = self.ping_test(ip_str) - - if 'error' not in ping_results: - ping_text = f""" -Packets: Sent = {ping_results['sent']}, Received = {ping_results['received']}, Lost = {ping_results['lost']} -Approximate round trip times: - Minimum = {ping_results['min']:.2f}ms - Maximum = {ping_results['max']:.2f}ms - Average = {ping_results['avg']:.2f}ms - """ - console.print(Panel(ping_text, title="[bold]📡 Connectivity Test[/bold]", border_style="green")) + + # 1. WHOIS Information (Enhanced Detail) + whois_data = data.get('ipwhois', {}) + if whois_data and 'error' not in whois_data: + whois_node = diag_tree.add("📜 [bold yellow]WHOIS Network Details[/bold yellow]") + whois_node.add(f"[dim]ASN:[/dim] [bold]{whois_data.get('asn', 'N/A')}[/bold] - {whois_data.get('asn_org', 'N/A')}") + whois_node.add(f"[dim]Owner:[/dim] {whois_data.get('org', 'N/A')}") + whois_node.add(f"[dim]Contact Email:[/dim] [dim]{whois_data.get('asn_contact', 'N/A')}[/dim]") + whois_node.add(f"[dim]Country:[/dim] {whois_data.get('asn_country', 'N/A')}") + whois_node.add(f"[dim]Range:[/dim] [dim]{whois_data.get('range', 'N/A')}[/dim]") + + # 2. Port Scan Results + ports = data.get('ports') + if isinstance(ports, dict): + open_ports = [str(p) for p, is_open in ports.items() if is_open] + + port_node = diag_tree.add(f"🔍 [bold #00ffff]Port Scan[/bold #00ffff] ([green]{len(open_ports)} OPEN[/green] / [red]{len(ports)-len(open_ports)} CLOSED[/red])") + if open_ports: + port_node.add(f"[green]Open Ports:[/green] [bold]{', '.join(open_ports)}[/bold]") + else: + port_node.add(f"[dim]Result:[/dim] [yellow]No open common ports detected.[/yellow]") + elif ports is None: + diag_tree.add(f"🔍 [dim]Port Scan:[/dim] Not executed ([bold yellow]Use -p[/bold yellow])") + + + # 3. Ping Test + ping_results = data.get('ping') + if ping_results: + if 'error' not in ping_results: + ping_node = diag_tree.add(f"📡 [bold #00ffff]Connectivity Test[/bold #00ffff]") + ping_node.add(f"[dim]Avg Latency:[/dim] [yellow]{ping_results['avg']:.2f}ms[/yellow] ([dim]Min/Max: {ping_results['min']:.2f}/{ping_results['max']:.2f}ms[/dim])") + ping_node.add(f"[dim]Packet Loss:[/dim] [red]{ping_results['lost']}[/red]/[dim]{ping_results['sent']}[/dim]") + else: + diag_tree.add(f"📡 [bold red]Connectivity Test Failed:[/bold red] {ping_results['error']}") + elif ping_results is None: + diag_tree.add(f"📡 [dim]Ping Test:[/dim] Not executed ([bold yellow]Use -t[/bold yellow])") + + + layout["diagnostics"].update(Panel(diag_tree, title="[bold bright_magenta]-- Diagnostics & Extras --[/bold bright_magenta]", border_style="magenta", box=box.ROUNDED)) + + console.print(layout) - def _display_results_plain(self, data: Dict, show_ports: bool, show_ping: bool): + def _display_results_plain(self, data: Dict): """Display results in plain text format""" print("\n" + "="*60) print("IP INTELLIGENCE REPORT") @@ -356,33 +429,36 @@ def _display_results_plain(self, data: Dict, show_ports: bool, show_ping: bool): if ip_api.get('status') == 'success': print(f"\nIP Address: {data.get('ip')}") print(f"Hostname: {data.get('reverse_dns', 'N/A')}") - print(f"\n📍 LOCATION") + print(f"\nLOCATION") print(f" Country: {ip_api.get('country')} ({ip_api.get('countryCode')})") - print(f" Region: {ip_api.get('regionName')}") print(f" City: {ip_api.get('city')}") - print(f" ZIP: {ip_api.get('zip')}") - print(f" Timezone: {ip_api.get('timezone')}") - print(f" Coordinates: {ip_api.get('lat')}, {ip_api.get('lon')}") - - print(f"\n🌐 NETWORK") + print(f"\nNETWORK") print(f" ISP: {ip_api.get('isp')}") print(f" Organization: {ip_api.get('org')}") print(f" AS: {ip_api.get('as')}") + + abuse_data = data.get('abuse_ipdb', {}) + if abuse_data and 'error' not in abuse_data: + print(f"\nSECURITY (AbuseIPDB)") + print(f" Confidence Score: {abuse_data.get('abuseConfidenceScore', 0)}%") + + whois_data = data.get('ipwhois', {}) + if whois_data and 'error' not in whois_data: + print(f"\nWHOIS DATA") + print(f" ASN: {whois_data.get('asn')} - {whois_data.get('asn_org')}") - if show_ports: - print("\n🔍 PORT SCAN") - ports = self.check_ports(data.get('ip')) + ports = data.get('ports') + if ports: + print("\nPORT SCAN") for port, is_open in ports.items(): status = "OPEN" if is_open else "CLOSED" print(f" Port {port}: {status}") - if show_ping: - print("\n📡 CONNECTIVITY TEST") - ping_results = self.ping_test(data.get('ip')) - if 'error' not in ping_results: - print(f" Sent: {ping_results['sent']}, Received: {ping_results['received']}") - print(f" Avg latency: {ping_results['avg']:.2f}ms") - + ping_results = data.get('ping') + if ping_results and 'error' not in ping_results: + print("\nCONNECTIVITY TEST") + print(f" Avg latency: {ping_results['avg']:.2f}ms") + print("\n" + "="*60) def export_json(self, data: Dict, filename: str): @@ -392,20 +468,45 @@ def export_json(self, data: Dict, filename: str): print(f"✓ Data exported to {filename}") def export_csv(self, data: Dict, filename: str): - """Export results to CSV""" - import csv + """Export results to CSV (Expanded fields)""" ip_api = data.get('ip_api', {}) + ipinfo = data.get('ipinfo', {}) + abuse_data = data.get('abuse_ipdb', {}) + + output_data = { + 'IP': data.get('ip'), + 'Timestamp': data.get('timestamp'), + 'Hostname_ReverseDNS': data.get('reverse_dns'), + 'Country': ip_api.get('country'), + 'City': ip_api.get('city'), + 'ISP': ip_api.get('isp'), + 'Organization': ip_api.get('org'), + 'AS': ip_api.get('as'), + 'Latitude': ip_api.get('lat'), + 'Longitude': ip_api.get('lon'), + 'IPinfo_Hostname': ipinfo.get('hostname'), + 'Abuse_Score': abuse_data.get('abuseConfidenceScore', 'N/A') if 'error' not in abuse_data else 'API Error', + 'Abuse_Reports': abuse_data.get('totalReports', 'N/A') if 'error' not in abuse_data else 'API Error', + } + + # Add Port Scan Data + ports = data.get('ports', {}) + for port, is_open in ports.items(): + output_data[f'Port_{port}_Status'] = 'OPEN' if is_open else 'CLOSED' + + # Add Ping Data + ping_results = data.get('ping', {}) + if 'error' not in ping_results: + output_data['Ping_Avg_ms'] = f"{ping_results['avg']:.2f}" + output_data['Ping_Min_ms'] = f"{ping_results['min']:.2f}" + output_data['Ping_Max_ms'] = f"{ping_results['max']:.2f}" + + with open(filename, 'w', newline='') as f: - writer = csv.writer(f) - writer.writerow(['Field', 'Value']) - writer.writerow(['IP', data.get('ip')]) - writer.writerow(['Hostname', data.get('reverse_dns')]) - writer.writerow(['Country', ip_api.get('country')]) - writer.writerow(['City', ip_api.get('city')]) - writer.writerow(['ISP', ip_api.get('isp')]) - writer.writerow(['Latitude', ip_api.get('lat')]) - writer.writerow(['Longitude', ip_api.get('lon')]) + writer = csv.DictWriter(f, fieldnames=output_data.keys()) + writer.writeheader() + writer.writerow(output_data) print(f"✓ Data exported to {filename}") @@ -419,7 +520,7 @@ def main(): parser.add_argument('ip', nargs='?', help='IP address to lookup (leave empty for your public IP)') parser.add_argument('-v', '--verbose', action='store_true', help='Verbose output') parser.add_argument('-p', '--ports', action='store_true', help='Scan common ports') - parser.add_argument('-t', '--ping', action='store_true', help='Perform ping test') + parser.add_argument('-t', '--ping', action='store_true', help='Perform system ping test') parser.add_argument('-a', '--abuse-key', help='AbuseIPDB API key for security analysis') parser.add_argument('-e', '--export', choices=['json', 'csv'], help='Export format') parser.add_argument('-o', '--output', help='Output filename') @@ -428,21 +529,23 @@ def main(): args = parser.parse_args() + # Initialize lookup tool lookup = IPLookup(verbose=args.verbose, timeout=args.timeout) if not args.no_banner: lookup.print_banner() - # Get IP address + # --- IP Detection --- if args.ip: ip = args.ip if RICH_AVAILABLE: - console.print(f"\n[bold cyan]Analyzing IP:[/bold cyan] {ip}\n") + console.print(f"\n[bold bright_white]TARGET ACQUIRED:[/bold bright_white] [bold yellow]{ip}[/bold yellow]\n") else: print(f"\nAnalyzing IP: {ip}\n") else: + # Detect public IP if RICH_AVAILABLE: - with Progress(SpinnerColumn(), TextColumn("[cyan]Detecting your public IP..."), console=console) as progress: + with Progress(SpinnerColumn(spinner_name="point"), TextColumn("[bold bright_white]DETECTING YOUR PUBLIC IP...[/bold bright_white]"), console=console) as progress: progress.add_task("", total=None) ip = lookup.get_public_ip() else: @@ -454,15 +557,22 @@ def main(): sys.exit(1) if RICH_AVAILABLE: - console.print(f"\n[bold green]✓ Your Public IP:[/bold green] {ip}\n") + console.print(f"\n[bold #00ff00]✓ PUBLIC IP DETECTED:[/bold #00ff00] [bold yellow]{ip}[/bold yellow]\n") else: print(f"\n✓ Your Public IP: {ip}\n") - # Gather and display data - data = lookup.aggregate_data(ip, args.abuse_key) - lookup.display_results(data, show_ports=args.ports, show_ping=args.ping) + # --- Gather Data (Parallelized) --- + data = lookup.aggregate_data( + ip, + args.abuse_key, + show_ports=args.ports, + show_ping=args.ping + ) + + # --- Display Results --- + lookup.display_results(data) - # Export if requested + # --- Export if requested --- if args.export: output_file = args.output or f"ip_lookup_{ip.replace('.', '_')}_{int(time.time())}.{args.export}" if args.export == 'json': @@ -471,8 +581,24 @@ def main(): lookup.export_csv(data, output_file) if RICH_AVAILABLE: - console.print("\n[dim]Thank you for using Ultimate IP Lookup Tool![/dim]") + console.print("\n[dim green]PROCESS COMPLETE. AWAITING NEXT COMMAND.[/dim green]") if __name__ == "__main__": - main() \ No newline at end of file + # Add exception handling for root/admin privilege issues (common for ping) + try: + main() + except PermissionError: + if RICH_AVAILABLE: + console.print("\n[bold white on red]FATAL ERROR: PERMISSION DENIED.[/bold white on red]", file=sys.stderr) + console.print("The system ping test requires [bold yellow]elevated privileges[/bold yellow] on this OS.", file=sys.stderr) + console.print("Try running the script with [bold yellow]'sudo python iplookup.py'[/bold yellow] if you use the -t/--ping flag.", file=sys.stderr) + else: + print("\nFATAL ERROR: Permission Denied. Ping test requires elevated privileges (sudo).", file=sys.stderr) + sys.exit(1) + except Exception as e: + if RICH_AVAILABLE: + console.print(f"\n[bold red]CRITICAL ERROR:[/bold red] {e}", file=sys.stderr) + else: + print(f"\nCRITICAL ERROR: {e}", file=sys.stderr) + sys.exit(1) \ No newline at end of file