port_scanner.py

import socket
import threading
from queue import Queue
import time
import argparse
from typing import List, Tuple

class PortScanner:
    COMMON_PORTS = {
        20: "FTP-DATA", 21: "FTP", 22: "SSH", 23: "TELNET",
        25: "SMTP", 53: "DNS", 80: "HTTP", 443: "HTTPS",
        3306: "MYSQL", 3389: "RDP"
    }
    def __init__(self, target: str, start_port: int = 1, end_port: int = 1024, threads: int = 50):
        self.target = target
        self.start_port = start_port
        self.end_port = end_port
        self.threads = threads
        self.queue = Queue()
        self.results = []
        
    def _is_port_open(self, port: int) -> Tuple[int, bool, str]:
        """Test if a port is open."""
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.settimeout(1)
                result = s.connect_ex((self.target, port))
                if result == 0:
                    try:
                        service = socket.getservbyport(port)
                    except:
                        service = "unknown"
                    return port, True, service
            return port, False, ""
        except:
            return port, False, ""
            
    def _worker(self):
        """Worker thread to process the port queue."""
        while True:
            port = self.queue.get()
            if port is None:
                break
            result = self._is_port_open(port)
            if result[1]:  # if port is open
                self.results.append(result)
            self.queue.task_done()
            
    def scan(self) -> List[Tuple[int, bool, str]]:
        """Start the port scanning process."""
        start_time = time.time()
        
        # Fill the queue with ports
        for port in range(self.start_port, self.end_port + 1):
            self.queue.put(port)
            
        # Start worker threads
        thread_list = []
        for _ in range(self.threads):
            t = threading.Thread(target=self._worker)
            t.start()
            thread_list.append(t)
            
        # Add sentinel values to stop threads
        for _ in range(self.threads):
            self.queue.put(None)
            
        # Wait for all threads to complete
        for t in thread_list:  # Changed this line from 'threading.Thread' to 'thread_list'
            t.join()
            
        end_time = time.time()
        self.scan_time = end_time - start_time
        return sorted(self.results)
    
    def _udp_scan(self, port: int) -> Tuple[int, bool, str]:
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
                s.settimeout(1)
                s.sendto(bytes(0), (self.target, port))
                try:
                    s.recvfrom(1024)
                    return port, True, "udp"
                except socket.timeout:
                    return port, False, ""
        except:
            return port, False, ""
    
    def _grab_banner(self, ip: str, port: int) -> str:
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.settimeout(2)
                s.connect((ip, port))
                banner = s.recv(1024).decode().strip()
                return banner
        except:
            return ""
    
    
def main():
    parser = argparse.ArgumentParser(description="Simple Python Port Scanner")
    parser.add_argument("target", help="Target host to scan")
    parser.add_argument("-s", "--start", type=int, default=1, help="Start port (default: 1)")
    parser.add_argument("-e", "--end", type=int, default=1024, help="End port (default: 1024)")
    parser.add_argument("-t", "--threads", type=int, default=50, help="Number of threads (default: 50)")
    args = parser.parse_args()

    # Create and run scanner
    scanner = PortScanner(args.target, args.start, args.end, args.threads)
    print(f"\nStarting scan on host {args.target}")
    
    try:
        results = scanner.scan()
        
        # Print results
        print(f"\nScan completed in {scanner.scan_time:.2f} seconds")
        print("\nOpen ports:")
        print("PORT\tSTATE\tSERVICE")
        print("-" * 30)
        
        for port, is_open, service in results:
            if is_open:
                print(f"{port}\topen\t{service}")
                
        print(f"\nScanned {args.end - args.start + 1} ports")
        print(f"Found {len(results)} open ports")
        
    except KeyboardInterrupt:
        print("\nScan interrupted by user")
    except socket.gaierror:
        print("\nHostname could not be resolved")
    except socket.error:
        print("\nCouldn't connect to server")

if __name__ == "__main__":
    main()