快捷導航

Python利用socket實現(xiàn)多進程的端口掃描器

更新時間：2022年12月06日 14:26:12 作者：Sir 老王

作為開發(fā)人員經(jīng)常需要查看服務(wù)的端口開啟狀態(tài)判斷服務(wù)是否宕機。特別是部署的服務(wù)比較多的情況下，可能存在幾個甚至幾十個服務(wù)端口的占用。所以本文將利用socket實現(xiàn)多進程的端口掃描器，需要的可以參考一下

作為開發(fā)人員經(jīng)常需要查看服務(wù)的端口開啟狀態(tài)判斷服務(wù)是否宕機。

特別是部署的服務(wù)比較多的情況下，可能存在幾個甚至幾十個服務(wù)端口的占用，于是我利用socket不斷向服務(wù)發(fā)送請求的方式來判斷端口服務(wù)是否已經(jīng)完成開啟。

其中加入多進程的調(diào)用方式來提高端口掃描的效率，供大家參考！

首先，我們將需要的python模塊全部導入到我們的代碼塊中，若是沒有安裝的模塊使用pip的當時安裝一下即可。

#?Importing?the?socket?module.
import?socket

#?Importing?the?datetime?module?from?the?datetime?package.
from?datetime?import?datetime

#?It's?a?shortcut?for?`from?multiprocessing?import?Pool`
from?multiprocessing.dummy?import?Pool

#?It's?a?shortcut?for?`from?loguru?import?logger`
from?loguru?import?logger

然后，創(chuàng)建一個端口掃描類PortScanner來完成對整個業(yè)務(wù)邏輯的處理，另外，封裝到類中也便于大家參考和修改。

class?PortScanner:
????def?__init__(self):
????????"""
????????A?constructor.?It?is?called?when?an?object?is?created?from?a?class?and?it?allows?the?class?to?initialize?the
????????attributes?of?a?class.
????????"""
????????super(PortScanner,?self).__init__()
????????self.remote_ip?=?None
????????self.ports?=?[]

????def?scanner(self,?port):
????????"""
????????It?scans?the?port.

????????:param?port:?The?port?you?want?to?scan
????????"""
????????try:
????????????socket_?=?socket.socket(socket.AF_INET,?socket.SOCK_STREAM)
????????????result_?=?socket_.connect_ex((self.remote_ip,?port))
????????????if?result_?==?0:
????????????????logger.info('地址:{}?端口:{}?已成功開啟！'.format(self.remote_ip,?port))
????????????else:
????????????????logger.info('地址:{}?端口:{}?未開啟！'.format(self.remote_ip,?port))
????????except?Exception?as?e:
????????????logger.error('端口掃描出現(xiàn)異常！')
????????finally:
????????????socket_.close()

????def?start(self):
????????"""
????????It?starts?the?game.
????????"""
????????remote_server?=?input("輸入要掃描的主機地址(127.0.0.1):")
????????if?remote_server.strip()?==?'':
????????????remote_server?=?'127.0.0.1'
????????self.remote_ip?=?socket.gethostbyname(remote_server)
????????port_range?=?input("輸入要掃描的端口范圍(1,50000):")
????????scanner_ports?=?[]
????????if?port_range.strip()?==?'':
????????????port_range?=?'1,50000'
????????scanner_ports?=?[n?for?n?in?range(int(port_range.split(',')[0]),?int(port_range.split(',')[1]))]
????????socket.setdefaulttimeout(0.5)
????????start_?=?datetime.now()
????????pool?=?Pool(processes=10)
????????pool.map(self.scanner,?scanner_ports)
????????pool.close()
????????pool.join()
????????end_?=?datetime.now()
????????logger.info('所有端口掃描已完成，總共耗時：{}'.format(str(end_?-?start_)))

使用python模塊中的main函數(shù)調(diào)用整個端口掃描器執(zhí)行掃描任務(wù)。

#?It's?a?common?idiom?to?determine?if?the?script?is?being?run?directly?or?being?imported.
if?__name__?==?'__main__':
????scanner_?=?PortScanner()
????scanner_.start()