怎么样制作80端口扫描器，批处理那种的。谢谢~

Posted 2023-04-30

会的加我的QQ:1573884655 也可以直接给方法，先谢谢了。。~ 批处理那种的啊。。网上找不到。！ 会制作的朋友，真的感谢您能帮个忙。！

参考技术A @echo
off
setlocal
enabledelayedexpansion
title
端口扫描器　by
风旋碧浪
color
4f&MODE
con:
COLS=50
LINES=12
echo.&echo
开始扫描，需要时间，请稍候……
(echo.&echo
端口号
进程名称
echo.&echo
TCP协议:
)>>端口查看.txt
echo.&echo
正在扫描TCP协议……
for
/f
"skip=4
tokens=2,5"
%%a
in
('netstat
-ano
-p
TCP')
do
(
for
/f
"tokens=2
delims=:"
%%A
in
("%%a")
do
set
TCP端口=%%A
for
/f
"skip=2
delims=,
tokens=1"
%%A
in
('tasklist
/fi
"PID
eq
%%b"
/FO
CSV')
do
set
TCP进程=%%~A
echo
!TCP端口!
!TCP进程!>>端口查看.txt
)
echo
UDP协议:
>>端口查看.txt
echo.&echo
正在扫描UDP协议……
for
/f
"skip=4
tokens=2,4"
%%a
in
('netstat
-ano
-p
UDP')
do
(
for
/f
"tokens=2
delims=:"
%%A
in
("%%a")
do
set
UDP端口=%%A
for
/f
"skip=2
delims=,
tokens=1"
%%A
in
('tasklist
/fi
"PID
eq
%%b"
/FO
CSV')
do
set
UDP进程=%%~A
echo
!UDP端口!
!UDP进程!>>端口查看.txt
)
echo.&echo.&echo
扫描结束！请查看《端口查看.txt》文件！
pause>nul

制作快速端口扫描器

【中文标题】制作快速端口扫描器

【英文标题】：Making a Fast Port Scanner

【发布时间】：2014-11-28 06:19:57

【问题描述】：

所以我在 python 中做一个端口扫描器...

import socket
ip = "External IP"
s = socket.socket(2, 1) #socket.AF_INET, socket.SOCK_STREAM

def porttry(ip, port):
    try:
        s.connect((ip, port))
        return True
    except:
        return None

for port in range(0, 10000):
    value = porttry(ip, port)
    if value == None:
        print("Port not opened on %d" % port)
    else:
        print("Port opened on %d" % port)
        break
raw_input()

但这太慢了，我想在一段时间没有返回任何东西后，能够以某种方式关闭或破坏代码。

【参考方案1】：

可以使用 threading.Thread 和 threading.Condition 来同步端口检查和产生新线程。

脚本示例用法：

python port_scan.py google.com 70 90
Checking 70 - 80
Checking 80 - 84
Checking 84 - 90
Found active port 80
Checking 90 - 91
Checking 91 - 94
All threads started ...

port_scan.py:

# import pdb
import socket, threading
from traceback import print_exc


class AllThreadsStarted(Exception): pass


class IPv4PortScanner(object):
    def __init__(self, domain, timeout=2.0, port_range=(1024, 65535), threadcount=10):
        self.domain               = domain
        self.timeout              = timeout
        self.port_range           = port_range
        self.threadcount          = threadcount
        self._lock                = threading.Lock()
        self._condition           = threading.Condition(self._lock)
        self._ports_active        = []
        self._ports_being_checked = []

        self._next_port = self.port_range[0]

    def check_port_(self, port):
        "If connects then port is active"
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.settimeout(self.timeout)
        try:
            sock.connect((self.domain, port))
            with self._lock:
                self._ports_active.append(port)
            print ("Found active port ".format(port))
            sock.close()
        except socket.timeout, ex:
            return
        except:
            print_exc()
            # pdb.set_trace()

    def check_port(self, port):
        "updates self._ports_being_checked list on exit of this method"
        try:
            self.check_port_(port)
        finally:
            self._condition.acquire()
            self._ports_being_checked.remove(port)
            self._condition.notifyAll()
            self._condition.release()

    def start_another_thread(self):
        if self._next_port > self.port_range[1]:
            raise AllThreadsStarted()
        port             = self._next_port
        self._next_port += 1
        t = threading.Thread(target=self.check_port, args=(port,))
        # update books
        with self._lock:
            self._ports_being_checked.append(port)
        t.start()

    def run(self):
        try:
            while True:
                self._condition.acquire()
                while len(self._ports_being_checked) >= self.threadcount:
                    # we wait for some threads to complete the task
                    self._condition.wait()
                slots_available = self.threadcount - len(self._ports_being_checked)
                self._condition.release()
                print ("Checking  - ".format(self._next_port, self._next_port+slots_available))
                for i in xrange(slots_available):
                    self.start_another_thread()
        except AllThreadsStarted, ex:
            print ("All threads started ...")
        except:
            print_exc()


if __name__ == "__main__":
    import sys
    domain  = sys.argv[1]
    port_s  = int(sys.argv[2])
    port_e  = int(sys.argv[3])
    scanner = IPv4PortScanner(domain=domain, port_range=(port_s, port_e))
    scanner.run()

【参考方案2】：

以下端口扫描器在顶部定义了一些常量，您可以根据需要进行修改：

目的 -- 命令行帮助消息 PORTS -- 您要扫描的端口范围 POOL_SIZE -- 要扫描的进程数 TIMEOUT -- 等待服务器连接的时间

请根据您的要求随意调整。也许添加一些命令行参数？

#! /usr/bin/env python3
import argparse
import collections
import itertools
import multiprocessing
import operator
import socket

PURPOSE = 'Scan for open ports on a computer.'
PORTS = range(1 << 16)
POOL_SIZE = 1 << 8
TIMEOUT = 0.01


def main():
    """Get computer to scan, connect with process pool, and show open ports."""
    parser = argparse.ArgumentParser(description=PURPOSE)
    parser.add_argument('host', type=str, help='computer you want to scan')
    host = parser.parse_args().host
    with multiprocessing.Pool(POOL_SIZE, socket.setdefaulttimeout, [TIMEOUT]) \
            as pool:
        results = pool.imap_unordered(test, ((host, port) for port in PORTS))
        servers = filter(operator.itemgetter(0), results)
        numbers = map(operator.itemgetter(1), servers)
        ordered = sorted(numbers)
    print(f'Ports open on host:', *format_ports(ordered), sep='\n    ')


field_names = 'family', 'socket_type', 'protocol', 'canon_name', 'address'
AddressInfo = collections.namedtuple('AddressInfo', field_names)
del field_names


def test(address):
    """Try connecting to the server and return whether or not it succeeded."""
    host, port = address
    for info in itertools.starmap(AddressInfo, socket.getaddrinfo(host, port)):
        try:
            probe = socket.socket(info.family, info.socket_type, info.protocol)
        except OSError:
            pass
        else:
            try:
                probe.connect(info.address)
            except OSError:
                pass
            else:
                probe.shutdown(socket.SHUT_RDWR)
                return True, port
            finally:
                probe.close()
    return False, port


def format_ports(ports):
    """Convert port numbers into strings and show all associated services."""
    if ports:
        for port in ports:
            try:
                service = socket.getservbyport(port)
            except OSError:
                service = '?'
            yield f'port:<5 = service'
    else:
        yield 'None'


if __name__ == '__main__':
    main()

【参考方案3】：

除了设置套接字超时，您还可以应用多线程技术来加速进程。当您有 N 个要扫描的端口时，它最多会快 N 倍。

# This script runs on Python 3
import socket, threading


def TCP_connect(ip, port_number, delay, output):
    TCPsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    TCPsock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    TCPsock.settimeout(delay)
    try:
        TCPsock.connect((ip, port_number))
        output[port_number] = 'Listening'
    except:
        output[port_number] = ''



def scan_ports(host_ip, delay):

    threads = []        # To run TCP_connect concurrently
    output =          # For printing purposes

    # Spawning threads to scan ports
    for i in range(10000):
        t = threading.Thread(target=TCP_connect, args=(host_ip, i, delay, output))
        threads.append(t)

    # Starting threads
    for i in range(10000):
        threads[i].start()

    # Locking the main thread until all threads complete
    for i in range(10000):
        threads[i].join()

    # Printing listening ports from small to large
    for i in range(10000):
        if output[i] == 'Listening':
            print(str(i) + ': ' + output[i])



def main():
    host_ip = input("Enter host IP: ")
    delay = int(input("How many seconds the socket is going to wait until timeout: "))   
    scan_ports(host_ip, delay)

if __name__ == "__main__":
    main()

【讨论】：

SO_REUSEADDR 仅在侦听套接字时有用。在这个例子中只会造成混乱

10000 导致错误 - 打开的文件太多，因此减少范围（1000）对我有用。

【参考方案4】：

这是一个快速简单的端口扫描器，它可以在 180 秒内扫描 100000 个端口：

import threading
import socket

target = 'pythonprogramming.net'
#ip = socket.gethostbyname(target)

def portscan(port):

    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(0.5)# 

    try:
        con = s.connect((target,port))

        print('Port :',port,"is open.")

        con.close()
    except: 
        pass
r = 1 
for x in range(1,100): 

    t = threading.Thread(target=portscan,kwargs='port':r) 

    r += 1     
    t.start() 

【讨论】：

如果你有x，为什么要使用r变量，它从1开始，每次迭代递增？

【参考方案5】：

这应该会快一点。

#-*-coding:utf8;-*-
#qpy:3
#qpy:console

import socket
import os

# This is used to set a default timeout on socket
# objects.
DEFAULT_TIMEOUT = 0.5

# This is used for checking if a call to socket.connect_ex
# was successful.
SUCCESS = 0

def check_port(*host_port, timeout=DEFAULT_TIMEOUT):
    ''' Try to connect to a specified host on a specified port.
    If the connection takes longer then the TIMEOUT we set we assume
    the host is down. If the connection is a success we can safely assume
    the host is up and listing on port x. If the connection fails for any
    other reason we assume the host is down and the port is closed.'''

    # Create and configure the socket.
    sock = socket.socket()
    sock.settimeout(timeout)

    # the SO_REUSEADDR flag tells the kernel to reuse a local 
    # socket in TIME_WAIT state, without waiting for its natural
    # timeout to expire.
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

    # Like connect(address), but return an error indicator instead
    # of raising an exception for errors returned by the C-level connect() 
    # call (other problems, such as “host not found,” can still raise exceptions). 
    # The error indicator is 0 if the operation succeeded, otherwise the value of 
    # the errnovariable. This is useful to support, for example, asynchronous connects.
    connected = sock.connect_ex(host_port) is SUCCESS

    # Mark the socket closed. 
    # The underlying system resource (e.g. a file descriptor)
    # is also closed when all file objects from makefile() are closed.
    # Once that happens, all future operations on the socket object will fail. 
    # The remote end will receive no more data (after queued data is flushed).
    sock.close()

    # return True if port is open or False if port is closed.
    return connected


con = check_port('www.google.com', 83)
print(con)

【参考方案6】：

socket.setdefualttimeout（时间）

用于在特定时间继续尝试连接端口...当您发送请求并且设置了 2 秒超时，因此它将尝试连接端口 2 秒...如果没有该端口在 2 秒内响应....它将被视为死端口

【参考方案7】：

socket.setdefaulttimeout(0.5) 这将使程序更快！

【讨论】：

这个答案是一个好的开始，但我认为解释为什么它会加速程序会很有用，也许有一些参考资料。

【参考方案8】：

考虑使用 socket.setdefaulttimeout(timeout) 设置超时而不是 for 循环。

【参考方案9】：

我认为这个 sn-p 可以帮助你：http://www.coderholic.com/python-port-scanner/