Author: blackcat

雖然坊間有很多port scanning 的software 可以用，但基於課習，自己寫了段 script 去查看自己部機有咩port 係開左，
為了可以快一點，我把 timeout set 了一秒，連不到就試下一個port，
但結都是很慢，如果想再快一點，可能要加thread 之類，下次有空再加

import socket

host = "192.168.1.1"

def pscan(ip, port):
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(1) # to fasten the scanning speed
        s.connect((ip, port))
        s.close()
        return True
    except:
        return False
    

for i in range(1, 65535, 1):
    if pscan(host, i):
        print("{}:{} is opened".format(host, i))
    if i%100 == 0:
        print("scanned {} ports".format(i)) # print the progress

第二版本，一個用了 thread 去 scan 開了的port ，感覺良好。

import os
import sys
import threading
import time
import queue
import random
import socket
import ipaddress

DEBUG=False

class Task:
    def __init__(self, ip, port, sock_type=socket.SOCK_STREAM):
        self.ip = ip
        self.port = port
        self.sock_type = sock_type

class Consumer(threading.Thread):
    def __init__(self, queue, name="consumer"):
        threading.Thread.__init__(self)
        self.name = name
        self.running = True
        self.queue = queue
        self.doneTaskCount = 0
    

    def run(self):
        while self.queue.empty() == False:
            task = self.queue.get();
            try:
                s = socket.socket(socket.AF_INET, task.sock_type)
                s.settimeout(1)
                s.connect((task.ip, task.port))
                print("{}:{} is opened".format(task.ip, task.port))
                s.close()
            except:
                pass
        #print("{} done".format(self.name))

class Monitor(threading.Thread):
    def __init__(self, queue, report_int=5, name="monitor"):
        threading.Thread.__init__(self)
        self.queue = queue
        self.report_int = report_int

    def run(self):
        while self.queue.empty() == False:
            print("remaining port: {}".format(self.queue.qsize()))
            time.sleep(self.report_int)
        # print("{} done".format(self.name))
      
          
# main
def main(ip_list, start_port=1, end_port=65535, thread_count=1000):
    print("starting main")
    consumer_count = thread_count
    
    Q = queue.Queue()
    
    # queue all target ip and port
    for ip in ip_list:
        if DEBUG:
            print("putting {} {} {}".format(ip, start_port, end_port))
        for i in range(start_port, end_port+1):
            task = Task(str(ip), i)
            Q.put(task)


    print("starting Monitor")
    monitor = Monitor(Q)
    monitor.start()

    print("start scanning with {} threads".format(consumer_count))

    consumer_arr = []
    for i in range(consumer_count):
        consumer_arr.append(Consumer(Q, "Consumer {}".format(i)))
        consumer_arr[i].start()
    
    
    # wait for finish
    for i in range(consumer_count):
        consumer_arr[i].join()
    monitor.join()    
    print("done. cleaning up.")
  
def help():
    print("{} <ip or subnet of the target> [range of the port]".format(sys.argv[0]))
    print("eg:")
    print("scan all of the opened port of the subnet 192.168.1.1 - 192.168.1.254")
    print("  {} 192.168.1.0/24".format(sys.argv[0]))
    print("")
    print("scan the opened port from 1 to 1024 of 192.168.1.1")
    print("  {} 192.168.1.1 1:1024".format(sys.argv[0]))

if __name__ == "__main__":
    ip_list = None
    start_port = 1
    end_port = 65535
    
    if len(sys.argv) < 2:
        help()
        exit()
    
    # parse ip
    try:
        net4 = ipaddress.ip_network(sys.argv[1])
        ips = list(net4.hosts())
        if len(ips) > 0 :
            ip_list = ips
        else:
            ip = ipaddress.ip_address(sys.argv[1])
            ip_list = [str(ip)]
    except Exception as e:
        print("invalid ip format")
        help()
        exit()

    # parse port
    if len(sys.argv) == 3:
        input_ip = sys.argv[1]
        port = sys.argv[2].split(":")
        if len(port) == 1:
            start_port = int(port[0])
            end_port = int(port[0])
        else:
            start_port = int(port[0])
            end_port = int(port[1])

    

    main(ip_list, start_port, end_port)