Advanced_Port_Scanner.py

#!/usr/bin/env python3
"""
Advanced Port Scanner with Banner Extraction
Enhanced version with comprehensive banner grabbing and threading
Author: Sergios9494
"""

import socket
import concurrent.futures
import sys
import argparse
import time
import json
import csv
from datetime import datetime
import threading
from queue import Queue

class PortScanner:
    def __init__(self, target, max_threads=100, timeout=3, verbose=False):
        self.target = target
        self.max_threads = max_threads
        self.timeout = timeout
        self.verbose = verbose
        self.results = []
        self.scan_start_time = None
        self.scan_end_time = None
        
        # Common service banners and detection patterns
        self.banner_patterns = {
            21: {'request': b'\r\n', 'service': 'FTP'},
            22: {'request': b'', 'service': 'SSH'},
            23: {'request': b'\r\n', 'service': 'Telnet'},
            25: {'request': b'EHLO test\r\n', 'service': 'SMTP'},
            53: {'request': b'', 'service': 'DNS'},
            80: {'request': b'GET / HTTP/1.1\r\nHost: ' + target.encode() + b'\r\n\r\n', 'service': 'HTTP'},
            110: {'request': b'', 'service': 'POP3'},
            135: {'request': b'', 'service': 'RPC'},
            139: {'request': b'', 'service': 'NetBIOS'},
            143: {'request': b'', 'service': 'IMAP'},
            443: {'request': b'GET / HTTP/1.1\r\nHost: ' + target.encode() + b'\r\n\r\n', 'service': 'HTTPS'},
            993: {'request': b'', 'service': 'IMAPS'},
            995: {'request': b'', 'service': 'POP3S'},
            3389: {'request': b'', 'service': 'RDP'},
            5432: {'request': b'', 'service': 'PostgreSQL'},
            3306: {'request': b'', 'service': 'MySQL'},
            6379: {'request': b'', 'service': 'Redis'},
            27017: {'request': b'', 'service': 'MongoDB'},
            8080: {'request': b'GET / HTTP/1.1\r\nHost: ' + target.encode() + b'\r\n\r\n', 'service': 'HTTP-Alt'},
            8443: {'request': b'GET / HTTP/1.1\r\nHost: ' + target.encode() + b'\r\n\r\n', 'service': 'HTTPS-Alt'},
            9200: {'request': b'', 'service': 'Elasticsearch'},
            5601: {'request': b'', 'service': 'Kibana'},
            3000: {'request': b'GET / HTTP/1.1\r\nHost: ' + target.encode() + b'\r\n\r\n', 'service': 'Node.js'},
            5000: {'request': b'GET / HTTP/1.1\r\nHost: ' + target.encode() + b'\r\n\r\n', 'service': 'Flask'},
            8000: {'request': b'GET / HTTP/1.1\r\nHost: ' + target.encode() + b'\r\n\r\n', 'service': 'HTTP-Alt'},
            9000: {'request': b'GET / HTTP/1.1\r\nHost: ' + target.encode() + b'\r\n\r\n', 'service': 'HTTP-Alt'}
        }

    def log(self, message, level="INFO"):
        """Log messages with timestamp"""
        timestamp = datetime.now().strftime("%H:%M:%S")
        if self.verbose or level == "ERROR":
            print(f"[{timestamp}] [{level}] {message}")

    def get_service_name(self, port):
        """Get service name for a port"""
        try:
            service = socket.getservbyport(port, 'tcp')
            return service
        except OSError:
            # Check our custom patterns
            if port in self.banner_patterns:
                return self.banner_patterns[port]['service']
            return f"Unknown-{port}"

    def extract_banner(self, sock, port):
        """Extract banner from open port"""
        try:
            sock.settimeout(self.timeout)
            
            # Check if we have a specific request pattern for this port
            if port in self.banner_patterns:
                request = self.banner_patterns[port]['request']
                if request:
                    sock.send(request)
            
            # Try to receive banner
            banner = sock.recv(4096).decode('utf-8', errors='ignore').strip()
            
            # Clean up banner
            if banner:
                # Remove common prefixes/suffixes
                banner = banner.replace('\r\n', '\n').replace('\r', '\n')
                banner_lines = [line.strip() for line in banner.split('\n') if line.strip()]
                return '\n'.join(banner_lines[:5])  # Limit to 5 lines
            
            return ""
            
        except socket.timeout:
            return ""
        except Exception as e:
            if self.verbose:
                self.log(f"Banner extraction error on port {port}: {str(e)}", "ERROR")
            return ""

    def scan_port(self, port):
        """Scan a single port"""
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(self.timeout)
            
            result = sock.connect_ex((self.target, port))
            
            if result == 0:
                # Port is open
                service = self.get_service_name(port)
                banner = self.extract_banner(sock, port)
                
                port_info = {
                    'port': port,
                    'service': service,
                    'banner': banner,
                    'status': 'Open',
                    'protocol': 'TCP',
                    'timestamp': datetime.now().isoformat()
                }
                
                self.log(f"Port {port} ({service}) is open")
                if banner and self.verbose:
                    self.log(f"Banner: {banner[:100]}...")
                
                return port_info
            else:
                return None
                
        except Exception as e:
            if self.verbose:
                self.log(f"Error scanning port {port}: {str(e)}", "ERROR")
            return None
        finally:
            try:
                sock.close()
            except:
                pass

    def scan_ports(self, port_list):
        """Scan multiple ports using threading"""
        self.log(f"Starting port scan on {self.target}")
        self.log(f"Scanning {len(port_list)} ports with {self.max_threads} threads")
        
        self.scan_start_time = time.time()
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_threads) as executor:
            # Submit all scan tasks
            future_to_port = {
                executor.submit(self.scan_port, port): port 
                for port in port_list
            }
            
            # Process completed scans
            completed = 0
            for future in concurrent.futures.as_completed(future_to_port):
                port = future_to_port[future]
                try:
                    result = future.result()
                    if result:
                        self.results.append(result)
                    completed += 1
                    
                    if not self.verbose:
                        print(f"\rProgress: {completed}/{len(port_list)} ports scanned", end='', flush=True)
                        
                except Exception as e:
                    self.log(f"Error processing port {port}: {str(e)}", "ERROR")
        
        if not self.verbose:
            print()  # New line after progress
        
        self.scan_end_time = time.time()
        duration = self.scan_end_time - self.scan_start_time
        
        self.log(f"Scan completed in {duration:.2f} seconds")
        self.log(f"Found {len(self.results)} open ports")

    def print_results(self):
        """Print scan results in a formatted table"""
        if not self.results:
            print("No open ports found.")
            return
        
        print("\n" + "="*120)
        print("PORT SCAN RESULTS")
        print("="*120)
        print(f"{'Port':<8} {'Service':<20} {'Status':<10} {'Banner':<60}")
        print("-"*120)
        
        for result in sorted(self.results, key=lambda x: x['port']):
            port = result['port']
            service = result['service']
            status = result['status']
            banner = result['banner']
            
            # Truncate banner for display
            display_banner = banner[:60] + "..." if len(banner) > 60 else banner
            if not display_banner:
                display_banner = "No banner"
            
            print(f"{port:<8} {service:<20} {status:<10} {display_banner:<60}")
            
            # Show full banner if verbose
            if self.verbose and banner and len(banner) > 60:
                banner_lines = banner.split('\n')
                for line in banner_lines:
                    print(f"{'':<8} {'':<20} {'':<10} {line}")
        
        print("="*120)
        print(f"Total open ports: {len(self.results)}")
        if self.scan_start_time and self.scan_end_time:
            duration = self.scan_end_time - self.scan_start_time
            print(f"Scan duration: {duration:.2f} seconds")

    def save_results(self, filename, format_type='json'):
        """Save results to file"""
        if not self.results:
            print("No results to save.")
            return
        
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        
        if format_type == 'json':
            output_file = f"{filename}_{timestamp}.json"
            with open(output_file, 'w') as f:
                json.dump(self.results, f, indent=2)
            print(f"Results saved to {output_file}")
            
        elif format_type == 'csv':
            output_file = f"{filename}_{timestamp}.csv"
            with open(output_file, 'w', newline='') as f:
                writer = csv.writer(f)
                writer.writerow(['Port', 'Service', 'Status', 'Protocol', 'Banner', 'Timestamp'])
                for result in self.results:
                    writer.writerow([
                        result['port'], result['service'], result['status'], 
                        result['protocol'], result['banner'], result['timestamp']
                    ])
            print(f"Results saved to {output_file}")

def parse_port_range(port_range):
    """Parse port range string (e.g., '1-1000', '80,443,8080')"""
    ports = set()
    
    for part in port_range.split(','):
        part = part.strip()
        if '-' in part:
            start, end = map(int, part.split('-'))
            ports.update(range(start, end + 1))
        else:
            ports.add(int(part))
    
    return sorted(list(ports))

def main():
    parser = argparse.ArgumentParser(
        description='Advanced Port Scanner with Banner Extraction',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python3 port_scanner.py 192.168.1.1 --ports 1-1000
  python3 port_scanner.py example.com --ports 80,443,8080,8443 --threads 50
  python3 port_scanner.py 10.0.0.1 --ports 1-65535 --verbose --output scan_results
  python3 port_scanner.py target.com --ports 1-1000 --format csv --timeout 5
        """
    )
    
    parser.add_argument('target', help='Target host (IP address or hostname)')
    parser.add_argument('--ports', '-p', default='1-1000', 
                       help='Port range to scan (e.g., 1-1000, 80,443,8080) (default: 1-1000)')
    parser.add_argument('--threads', '-t', type=int, default=100, 
                       help='Maximum number of threads (default: 100)')
    parser.add_argument('--timeout', type=int, default=3, 
                       help='Timeout for connections in seconds (default: 3)')
    parser.add_argument('--verbose', '-v', action='store_true', 
                       help='Enable verbose output')
    parser.add_argument('--output', '-o', help='Output filename (without extension)')
    parser.add_argument('--format', choices=['json', 'csv'], default='json',
                       help='Output format (default: json)')
    
    args = parser.parse_args()
    
    try:
        # Parse port range
        port_list = parse_port_range(args.ports)
        
        # Create scanner instance
        scanner = PortScanner(
            target=args.target,
            max_threads=args.threads,
            timeout=args.timeout,
            verbose=args.verbose
        )
        
        # Perform port scan
        scanner.scan_ports(port_list)
        
        # Display results
        scanner.print_results()
        
        # Save results if requested
        if args.output:
            scanner.save_results(args.output, args.format)
            
    except KeyboardInterrupt:
        print("\n[!] Scan interrupted by user")
        sys.exit(1)
    except Exception as e:
        print(f"[!] Error: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    main()