Welcome, guest | Sign In | My Account | Store | Cart

Python portscanners (Python recipe) by Foo Bear
ActiveState Code (http://code.activestate.com/recipes/286240/)

Those are a couple of multithreaded portscanners, the second one use the Queue module.

Python, 128 lines
Download
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
# a simple portscanner with multithreading

import socket as sk
import sys
import threading

MAX_THREADS = 50

def usage():
    print "\npyScan 0.1"
    print "usage: pyScan <host> [start port] [end port]"
    
class Scanner(threading.Thread):
    def __init__(self, host, port):
        threading.Thread.__init__(self)
        # host and port
        self.host = host
        self.port = port
        # build up the socket obj
        self.sd = sk.socket(sk.AF_INET, sk.SOCK_STREAM)

    def run(self):
        try:
            # connect to the given host:port
            self.sd.connect((self.host, self.port))
            print "%s:%d OPEN" % (self.host, self.port)
            self.sd.close()
        except: pass

class pyScan:
    def __init__(self, args=[]):
        # arguments vector
        self.args = args
        # start port and end port
        self.start, self.stop = 1, 1024
        # host name
        self.host = ""

        # check the arguments
        if len(self.args) == 4:
            self.host = self.args[1]
            try:
                self.start = int(self.args[2])
                self.stop = int(self.args[3])
            except ValueError:
                usage()
                return
            if self.start > self.stop:
                usage()
                return
        elif len(self.args) == 2:
            self.host = self.args[1]
        else:
            usage()
            return

        try:
            sk.gethostbyname(self.host)
        except:
            print "hostname '%s' unknown" % self.host
        self.scan(self.host, self.start, self.stop)

    def scan(self, host, start, stop):
        self.port = start
        while self.port <= stop:
            while threading.activeCount() < MAX_THREADS:
                Scanner(host, self.port).start()
                self.port += 1
        
if __name__ == "__main__":
    pyScan(sys.argv)

#############################################################

# a simple portscanner with multithreading
# QUEUE BASED VERSION

import socket
import sys
import threading, Queue

MAX_THREADS = 50

class Scanner(threading.Thread):
    def __init__(self, inq, outq):
        threading.Thread.__init__(self)
        self.setDaemon(1)
        # queues for (host, port)
        self.inq = inq
        self.outq = outq

    def run(self):
        while 1:
            host, port = self.inq.get()
            sd = socket.socket(socket.AF_INET, socket.SOCK_STREAM)        
            try:
                # connect to the given host:port
                sd.connect((host, port))
            except socket.error:
                # set the CLOSED flag
                self.outq.put((host, port, 'CLOSED'))
            else:
                self.outq.put((host, port, 'OPEN'))
                sd.close()

def scan(host, start, stop, nthreads=MAX_THREADS):
    toscan = Queue.Queue()
    scanned = Queue.Queue()

    scanners = [Scanner(toscan, scanned) for i in range(nthreads)]
    for scanner in scanners:
        scanner.start()

    hostports = [(host, port) for port in xrange(start, stop+1)]
    for hostport in hostports:
        toscan.put(hostport)

    results = {}
    for host, port in hostports:
        while (host, port) not in results:
            nhost, nport, nstatus = scanned.get()
            results[(nhost, nport)] = nstatus
        status = results[(host, port)]
        if status <> 'CLOSED':
            print '%s:%d %s' % (host, port, status)

if __name__ == '__main__':
    scan('localhost', 0, 1024)

Use them at your own risk, portscanning is not always friendly accepted :)

Tags: network

2 comments

BJ Swope 12 years, 5 months ago  # | flag

I'm using Python 2.6.6 and ran into an infinite loop with the first script.

The nested while loops in the scan function loops forever because my scans never utilized more than ~4 threads. I changed the second "while" to an "if" and the script worked beautifully.

Mac 12 years, 3 months ago  # | flag

Checkout this link. It's a simple port scanner developed in python. http://malhar2010.blogspot.com/2012/01/how-to-create-simple-port-scanner-in.html